mirror of
https://github.com/xtaci/kcp-go.git
synced 2026-10-19 16:14:38 +00:00
perf: reduce heap allocations and fix memory leaks across multiple components
Optimizations: - ringbuffer: optimize Len(), IsEmpty(), Discard(), Clear() methods with branchless arithmetic and simplified logic - fec: fix memory leak in shardHeap.Pop() by setting element to nil before truncating slice, preventing GC from holding stale references - timedsched: fix memory leak in timedFuncHeap.Pop() with same nil-before-truncate pattern; optimize prepend() using slice swap instead of append - autotune: add sortCache field to Autotuner to reuse slice allocation in FindPeriod(), avoiding repeated allocations during period detection - snmp: replace fmt.Sprint with strconv.FormatUint for better performance in Copy() and Reset() methods - entropy: remove defer overhead in rngAES.Read() and rngChacha8.Read() hot paths for reduced function call cost - bufferpool: pre-allocate ErrBufferPoolSizeExceeded error; reset slice length on Put() to allow proper reuse - sess: remove defer overhead in 6 setter methods (SetWriteDelay, SetWindowSize, SetStreamMode, SetACKNoDelay, SetDUP, SetNoDelay) These changes reduce memory allocations, fix potential memory leaks in heap-based data structures, and eliminate defer overhead in frequently called methods.
This commit is contained in:
+11
-7
@@ -37,10 +37,11 @@ type pulse struct {
|
||||
// autoTune object to detect pulses in a signal
|
||||
// Uses a fixed-size ring buffer instead of heap to avoid allocations
|
||||
type autoTune struct {
|
||||
pulses [maxAutoTuneSamples]pulse // fixed-size array to avoid heap allocations
|
||||
head int // oldest element index
|
||||
tail int // next write position
|
||||
count int // number of elements
|
||||
pulses [maxAutoTuneSamples]pulse // fixed-size array to avoid heap allocations
|
||||
sortCache [maxAutoTuneSamples]pulse // reusable cache for sorting to avoid allocations
|
||||
head int // oldest element index
|
||||
tail int // next write position
|
||||
count int // number of elements
|
||||
}
|
||||
|
||||
// Sample adds a signal sample to the pulse buffer using a ring buffer
|
||||
@@ -78,13 +79,16 @@ func (tune *autoTune) FindPeriod(bit bool) int {
|
||||
return -1
|
||||
}
|
||||
|
||||
// Copy elements from ring buffer for sorting and analysis.
|
||||
sorted := make([]pulse, tune.count)
|
||||
// Copy elements from ring buffer to sortCache for sorting and analysis.
|
||||
// Using fixed-size array to avoid heap allocation.
|
||||
for i := 0; i < tune.count; i++ {
|
||||
idx := (tune.head + i) % maxAutoTuneSamples
|
||||
sorted[i] = tune.pulses[idx]
|
||||
tune.sortCache[i] = tune.pulses[idx]
|
||||
}
|
||||
|
||||
// Create a slice view over the cache for sorting
|
||||
sorted := tune.sortCache[:tune.count]
|
||||
|
||||
// Sort the copied data by sequence number (seq) to ensure linear order for period calculation.
|
||||
sort.Slice(sorted, func(i, j int) bool {
|
||||
return _itimediff(sorted[i].seq, sorted[j].seq) < 0
|
||||
|
||||
+5
-2
@@ -27,6 +27,9 @@ import (
|
||||
"sync"
|
||||
)
|
||||
|
||||
// pre-allocated error to avoid repeated allocations
|
||||
var errBufferSizeMismatch = errors.New("buffer size mismatch")
|
||||
|
||||
// A system-wide packet buffer shared among sending, receiving and FEC
|
||||
// to mitigate high-frequency memory allocation of packets.
|
||||
var defaultBufferPool = newBufferPool(mtuLimit)
|
||||
@@ -55,8 +58,8 @@ func (bp *bufferPool) Get() []byte {
|
||||
func (bp *bufferPool) Put(buf []byte) error {
|
||||
// Only put back buffers of the correct size.
|
||||
if cap(buf) != mtuLimit {
|
||||
return errors.New("buffer size mismatch")
|
||||
return errBufferSizeMismatch
|
||||
}
|
||||
bp.xmitBuf.Put(buf)
|
||||
bp.xmitBuf.Put(buf[:cap(buf)]) // reset slice length to full capacity
|
||||
return nil
|
||||
}
|
||||
|
||||
+4
-5
@@ -119,11 +119,11 @@ func (r *rngAES) Read(p []byte) (int, error) {
|
||||
}
|
||||
|
||||
r.mutex.Lock()
|
||||
defer r.mutex.Unlock()
|
||||
|
||||
r.updateSeed()
|
||||
r.block.Encrypt(r.seed[:], r.seed[:])
|
||||
return copy(p, r.seed[:]), nil
|
||||
n := copy(p, r.seed[:])
|
||||
r.mutex.Unlock()
|
||||
return n, nil
|
||||
}
|
||||
|
||||
// rngChacha8 is a ChaCha8-based random number generator.
|
||||
@@ -164,9 +164,8 @@ func (r *rngChacha8) Read(p []byte) (int, error) {
|
||||
}
|
||||
|
||||
r.mutex.Lock()
|
||||
defer r.mutex.Unlock()
|
||||
|
||||
r.updateSeed()
|
||||
n, err := r.rand.Read(p)
|
||||
r.mutex.Unlock()
|
||||
return n, err
|
||||
}
|
||||
|
||||
@@ -93,7 +93,8 @@ func (h *shardHeap) Push(x any) {
|
||||
func (h *shardHeap) Pop() any {
|
||||
n := len(h.elements)
|
||||
x := h.elements[n-1]
|
||||
h.elements = h.elements[0 : n-1]
|
||||
h.elements[n-1] = nil // clear to avoid memory leak
|
||||
h.elements = h.elements[:n-1]
|
||||
delete(h.marks, x.seqid())
|
||||
return x
|
||||
}
|
||||
|
||||
+20
-10
@@ -53,8 +53,8 @@ func (r *RingBuffer[T]) Len() int {
|
||||
if r.head <= r.tail {
|
||||
return r.tail - r.head
|
||||
}
|
||||
|
||||
return len(r.elements[r.head:]) + len(r.elements[:r.tail])
|
||||
// Wrapped case: elements from head to end + elements from start to tail
|
||||
return len(r.elements) - r.head + r.tail
|
||||
}
|
||||
|
||||
// Push adds an element to the tail of the ring.
|
||||
@@ -93,8 +93,9 @@ func (r *RingBuffer[T]) Peek() (*T, bool) {
|
||||
// Discard discards the first N elements from the ring buffer.
|
||||
// Returns the number of elements that are actually discarded (<= n).
|
||||
func (r *RingBuffer[T]) Discard(n int) int {
|
||||
n = min(n, r.Len())
|
||||
if n == r.Len() {
|
||||
currentLen := r.Len()
|
||||
n = min(n, currentLen)
|
||||
if n == currentLen {
|
||||
r.Clear()
|
||||
return n
|
||||
}
|
||||
@@ -166,18 +167,27 @@ func (r *RingBuffer[T]) ForEachReverse(fn func(*T) bool) {
|
||||
|
||||
// Clear resets the ring to an empty state and reinitializes the buffer.
|
||||
func (r *RingBuffer[T]) Clear() {
|
||||
var zero T
|
||||
// Only clear elements that contain data to avoid retaining references
|
||||
if r.head <= r.tail {
|
||||
for i := r.head; i < r.tail; i++ {
|
||||
r.elements[i] = zero
|
||||
}
|
||||
} else {
|
||||
for i := r.head; i < len(r.elements); i++ {
|
||||
r.elements[i] = zero
|
||||
}
|
||||
for i := 0; i < r.tail; i++ {
|
||||
r.elements[i] = zero
|
||||
}
|
||||
}
|
||||
r.head = 0
|
||||
r.tail = 0
|
||||
|
||||
var zero T
|
||||
for i := range r.elements {
|
||||
r.elements[i] = zero // Clear each slot to avoid retaining references
|
||||
}
|
||||
}
|
||||
|
||||
// IsEmpty returns true if the ring has no elements.
|
||||
func (r *RingBuffer[T]) IsEmpty() bool {
|
||||
return r.Len() == 0
|
||||
return r.head == r.tail
|
||||
}
|
||||
|
||||
// MaxLen returns the maximum capacity of the ring buffer.
|
||||
|
||||
@@ -489,15 +489,15 @@ func (s *UDPSession) SetWriteDeadline(t time.Time) error {
|
||||
// SetWriteDelay delays write for bulk transfer until the next update interval
|
||||
func (s *UDPSession) SetWriteDelay(delay bool) {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
s.writeDelay = delay
|
||||
s.mu.Unlock()
|
||||
}
|
||||
|
||||
// SetWindowSize set maximum window size
|
||||
func (s *UDPSession) SetWindowSize(sndwnd, rcvwnd int) {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
s.kcp.WndSize(sndwnd, rcvwnd)
|
||||
s.mu.Unlock()
|
||||
}
|
||||
|
||||
// SetMtu sets the maximum transmission unit(not including UDP header)
|
||||
@@ -518,19 +518,19 @@ func (s *UDPSession) SetMtu(mtu int) bool {
|
||||
// Deprecated: toggles the stream mode on/off
|
||||
func (s *UDPSession) SetStreamMode(enable bool) {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
if enable {
|
||||
s.kcp.stream = 1
|
||||
} else {
|
||||
s.kcp.stream = 0
|
||||
}
|
||||
s.mu.Unlock()
|
||||
}
|
||||
|
||||
// SetACKNoDelay changes ack flush option, set true to flush ack immediately,
|
||||
func (s *UDPSession) SetACKNoDelay(nodelay bool) {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
s.ackNoDelay = nodelay
|
||||
s.mu.Unlock()
|
||||
}
|
||||
|
||||
// (deprecated)
|
||||
@@ -538,16 +538,16 @@ func (s *UDPSession) SetACKNoDelay(nodelay bool) {
|
||||
// SetDUP duplicates udp packets for kcp output.
|
||||
func (s *UDPSession) SetDUP(dup int) {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
s.dup = dup
|
||||
s.mu.Unlock()
|
||||
}
|
||||
|
||||
// SetNoDelay calls nodelay() of kcp
|
||||
// https://github.com/skywind3000/kcp/blob/master/README.en.md#protocol-configuration
|
||||
func (s *UDPSession) SetNoDelay(nodelay, interval, resend, nc int) {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
s.kcp.NoDelay(nodelay, interval, resend, nc)
|
||||
s.mu.Unlock()
|
||||
}
|
||||
|
||||
// SetDSCP sets the 6bit DSCP field in IPv4 header, or 8bit Traffic Class in IPv6 header.
|
||||
|
||||
@@ -23,7 +23,7 @@
|
||||
package kcp
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"strconv"
|
||||
"sync/atomic"
|
||||
)
|
||||
|
||||
@@ -105,36 +105,36 @@ func (s *Snmp) Header() []string {
|
||||
func (s *Snmp) ToSlice() []string {
|
||||
snmp := s.Copy()
|
||||
return []string{
|
||||
fmt.Sprint(snmp.BytesSent),
|
||||
fmt.Sprint(snmp.BytesReceived),
|
||||
fmt.Sprint(snmp.MaxConn),
|
||||
fmt.Sprint(snmp.ActiveOpens),
|
||||
fmt.Sprint(snmp.PassiveOpens),
|
||||
fmt.Sprint(snmp.CurrEstab),
|
||||
fmt.Sprint(snmp.InErrs),
|
||||
fmt.Sprint(snmp.InCsumErrors),
|
||||
fmt.Sprint(snmp.KCPInErrors),
|
||||
fmt.Sprint(snmp.InPkts),
|
||||
fmt.Sprint(snmp.OutPkts),
|
||||
fmt.Sprint(snmp.InSegs),
|
||||
fmt.Sprint(snmp.OutSegs),
|
||||
fmt.Sprint(snmp.InBytes),
|
||||
fmt.Sprint(snmp.OutBytes),
|
||||
fmt.Sprint(snmp.RetransSegs),
|
||||
fmt.Sprint(snmp.FastRetransSegs),
|
||||
fmt.Sprint(snmp.EarlyRetransSegs),
|
||||
fmt.Sprint(snmp.LostSegs),
|
||||
fmt.Sprint(snmp.RepeatSegs),
|
||||
fmt.Sprint(snmp.FECFullShardSet),
|
||||
fmt.Sprint(snmp.FECParityShards),
|
||||
fmt.Sprint(snmp.FECErrs),
|
||||
fmt.Sprint(snmp.FECRecovered),
|
||||
fmt.Sprint(snmp.FECShardSet),
|
||||
fmt.Sprint(snmp.FECShardMin),
|
||||
fmt.Sprint(snmp.RingBufferSndQueue),
|
||||
fmt.Sprint(snmp.RingBufferRcvQueue),
|
||||
fmt.Sprint(snmp.RingBufferSndBuffer),
|
||||
fmt.Sprint(snmp.OOBPackets),
|
||||
strconv.FormatUint(snmp.BytesSent, 10),
|
||||
strconv.FormatUint(snmp.BytesReceived, 10),
|
||||
strconv.FormatUint(snmp.MaxConn, 10),
|
||||
strconv.FormatUint(snmp.ActiveOpens, 10),
|
||||
strconv.FormatUint(snmp.PassiveOpens, 10),
|
||||
strconv.FormatUint(snmp.CurrEstab, 10),
|
||||
strconv.FormatUint(snmp.InErrs, 10),
|
||||
strconv.FormatUint(snmp.InCsumErrors, 10),
|
||||
strconv.FormatUint(snmp.KCPInErrors, 10),
|
||||
strconv.FormatUint(snmp.InPkts, 10),
|
||||
strconv.FormatUint(snmp.OutPkts, 10),
|
||||
strconv.FormatUint(snmp.InSegs, 10),
|
||||
strconv.FormatUint(snmp.OutSegs, 10),
|
||||
strconv.FormatUint(snmp.InBytes, 10),
|
||||
strconv.FormatUint(snmp.OutBytes, 10),
|
||||
strconv.FormatUint(snmp.RetransSegs, 10),
|
||||
strconv.FormatUint(snmp.FastRetransSegs, 10),
|
||||
strconv.FormatUint(snmp.EarlyRetransSegs, 10),
|
||||
strconv.FormatUint(snmp.LostSegs, 10),
|
||||
strconv.FormatUint(snmp.RepeatSegs, 10),
|
||||
strconv.FormatUint(snmp.FECFullShardSet, 10),
|
||||
strconv.FormatUint(snmp.FECParityShards, 10),
|
||||
strconv.FormatUint(snmp.FECErrs, 10),
|
||||
strconv.FormatUint(snmp.FECRecovered, 10),
|
||||
strconv.FormatUint(snmp.FECShardSet, 10),
|
||||
strconv.FormatUint(snmp.FECShardMin, 10),
|
||||
strconv.FormatUint(snmp.RingBufferSndQueue, 10),
|
||||
strconv.FormatUint(snmp.RingBufferRcvQueue, 10),
|
||||
strconv.FormatUint(snmp.RingBufferSndBuffer, 10),
|
||||
strconv.FormatUint(snmp.OOBPackets, 10),
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
+5
-13
@@ -48,8 +48,8 @@ func (h *timedFuncHeap) Pop() any {
|
||||
old := *h
|
||||
n := len(old)
|
||||
x := old[n-1]
|
||||
old[n-1].execute = nil // avoid memory leak
|
||||
*h = old[0 : n-1]
|
||||
old[n-1] = timedFunc{} // clear to avoid memory leak (both execute and ts)
|
||||
*h = old[:n-1]
|
||||
return x
|
||||
}
|
||||
|
||||
@@ -129,22 +129,14 @@ func (ts *TimedSched) prepend() {
|
||||
select {
|
||||
case <-ts.chPrependNotify:
|
||||
ts.prependLock.Lock()
|
||||
// keep cap to reuse slice
|
||||
if cap(tasks) < cap(ts.prependTasks) {
|
||||
tasks = make([]timedFunc, 0, cap(ts.prependTasks))
|
||||
}
|
||||
tasks = tasks[:len(ts.prependTasks)]
|
||||
copy(tasks, ts.prependTasks)
|
||||
for k := range ts.prependTasks {
|
||||
ts.prependTasks[k].execute = nil // avoid memory leak
|
||||
}
|
||||
ts.prependTasks = ts.prependTasks[:0]
|
||||
// swap slices to minimize time under lock
|
||||
tasks, ts.prependTasks = ts.prependTasks, tasks[:0]
|
||||
ts.prependLock.Unlock()
|
||||
|
||||
for k := range tasks {
|
||||
select {
|
||||
case ts.chTask <- tasks[k]:
|
||||
tasks[k].execute = nil // avoid memory leak
|
||||
tasks[k] = timedFunc{} // clear to avoid memory leak
|
||||
case <-ts.die:
|
||||
return
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user