github/kcp-go
4
0
Fork 0
mirror of https://github.com/xtaci/kcp-go.git synced 2026-10-19 16:14:38 +00:00
Code Issues Releases Wiki Activity

docs: improve code readability with structured comments and section headers

Browse Source 
- kcp.go: Group protocol constants by category (RTO, commands, probe flags,
  window/MTU defaults, protocol parameters) with descriptive comments.
  Document KCP struct fields with protocol semantics (RFC 6298/5681/6937).
  Add 6-phase description to flush() and annotate each phase.

- sess.go: Add section separators for session, packet input pipeline, listener,
  and public API. Document postProcess() 3-stage pipeline and kcpInput()
  packet demultiplexing logic.

- fec.go: Document FEC constants (header sizes, packet types, maxShardSets).
  Clarify fecDecoder's auto-tuning mechanism.

- crypt.go: Add section header for CFB-mode encryption. Document encrypt/decrypt
  dispatch and 8x loop unrolling optimization strategy.

- timedsched.go: Document two-stage pipeline architecture (prepend → sched)
  and explain why stages are separated.

- autotune.go: Expand FindPeriod() documentation with algorithm description
  and fix ASCII signal diagram alignment.

- ringbuffer.go: Document RINGBUFFER_MIN and RINGBUFFER_EXP constants.

- snmp.go: Fix typos (accmulated → accumulated, retransmited → retransmitted).

No code logic changes. All modifications are comments only.
This commit is contained in:
xtaci
2026-03-13 17:01:51 +08:00
parent 3f7d4c8b32
commit fec60d1ec6
8 changed files with 291 additions and 142 deletions
Download Patch File Download Diff File Expand all files Collapse all files
autotune.go
+34 -19
View File
@@ -26,16 +26,22 @@ import (
"sort"
)
const maxAutoTuneSamples = 258 // 256 + 2 extra for edge detection
const maxAutoTuneSamples = 258 // 256 + 2 extra samples for edge detection
// pulse represents a 0/1 signal with time sequence
// pulse represents a single sample in the signal stream: a boolean bit
// (data=true, parity=false) tagged with its FEC sequence number.
type pulse struct {
bit bool // 0 or 1
seq uint32 // sequence of the signal
bit bool // true = data shard, false = parity shard
seq uint32 // FEC sequence number of this packet
}
// autoTune object to detect pulses in a signal
// Uses a fixed-size ring buffer instead of heap to avoid allocations
// autoTune detects the repeating period of data and parity shards in the
// incoming FEC stream. This allows the decoder to auto-detect the peer's
// dataShards/parityShards configuration.
//
// Algorithm: collect recent samples into a ring buffer, sort by sequence,
// then scan for the first complete pulse (rising edge -> falling edge)
// of the target signal. The pulse width equals the shard count.
type autoTune struct {
pulses [maxAutoTuneSamples]pulse // fixed-size array to avoid heap allocations
sortCache [maxAutoTuneSamples]pulse // reusable cache for sorting to avoid allocations
@@ -58,21 +64,30 @@ func (tune *autoTune) Sample(bit bool, seq uint32) {
}
}
// Find a period for a given signal
// returns -1 if not found
// FindPeriod detects the period (pulse width) of the given signal type
// in the collected samples. Returns -1 if no valid period is found.
//
// For bit=true (data shards), it finds how many consecutive data shards
// appear before a parity shard, giving us the dataShards count.
// For bit=false (parity shards), it gives the parityShards count.
//
// Signal Level
// |
// 1.0 | _____ _____
// | | | | |
// 0.5 | _____ | | _____ | | _____
// | | | | | | || | | |
// 0.0 __|_____| |____| |__| || |__| |_____
// |
// |-----------------------------------------------------> Time
// A B C D E F G H I
// The detection works by finding a complete pulse in the sorted sequence:
// 1. Find the "left edge": where the signal transitions TO the target bit
// 2. Find the "right edge": where the signal transitions AWAY from the target bit
// 3. The period = rightEdge - leftEdge
//
// Example signal (bit=true looking for data period):
//
// Signal Level
// |
// 1.0 | _____ _____
// | | | | |
// 0.5 | _____ | | _____ | | _____
// | | | | | | || | | |
// 0.0 |_____| |____| |__| || |__| |_____
// |
// |-----------------------------------------------------> Time
// A B C D E F G H I
func (tune *autoTune) FindPeriod(bit bool) int {
// Need at least 3 samples to detect a period (rising and falling edges)
if tune.count < 3 {
crypt.go
+38 -19
View File
@@ -316,7 +316,20 @@ func (c *noneBlockCrypt) Decrypt(dst, src []byte) {
}
}
// packet encryption with local CFB mode
// -----------------------------------------------------------------------
// CFB-mode encryption/decryption
// -----------------------------------------------------------------------
//
// For block ciphers (non-AEAD), packets are encrypted using a local variant
// of CFB (Cipher Feedback) mode. The encrypt/decrypt functions below are
// hand-optimized with 8x loop unrolling to reduce data dependencies
// between consecutive block cipher calls.
//
// Two block sizes are supported:
// - 8-byte blocks (e.g. Blowfish, CAST5, DES, TEA, XTEA)
// - 16-byte blocks (e.g. AES, Twofish, SM4)
// encrypt dispatches to the appropriate block-size-specific CFB encryptor.
func encrypt(block cipher.Block, dst, src, buf []byte) {
switch block.BlockSize() {
case 8:
@@ -328,14 +341,15 @@ func encrypt(block cipher.Block, dst, src, buf []byte) {
}
}
// optimized encryption for the ciphers which works in 8-bytes
// encrypt8 performs CFB encryption for 8-byte block ciphers.
// Uses 8x loop unrolling for throughput optimization.
func encrypt8(block cipher.Block, dst, src, buf []byte) {
tbl := buf[:8]
block.Encrypt(tbl, initialVector)
n := len(src) >> 3 // len(src) / 8
n := len(src) >> 3 // number of full 8-byte blocks
base := 0
repeat := n >> 3 // n / 8
left := n & 7 // n % 8
repeat := n >> 3 // number of 8-block groups (64 bytes each)
left := n & 7 // remaining blocks after groups
for range repeat {
s := src[base:][0:64]
@@ -408,14 +422,15 @@ func encrypt8(block cipher.Block, dst, src, buf []byte) {
}
}
// optimized encryption for the ciphers which works in 16-bytes
// encrypt16 performs CFB encryption for 16-byte block ciphers.
// Uses 8x loop unrolling for throughput optimization.
func encrypt16(block cipher.Block, dst, src, buf []byte) {
tbl := buf[:16]
block.Encrypt(tbl, initialVector)
n := len(src) >> 4 // len(src) / 16
n := len(src) >> 4 // number of full 16-byte blocks
base := 0
repeat := n >> 3 // n / 8
left := n & 7 // n % 8
repeat := n >> 3 // number of 8-block groups (128 bytes each)
left := n & 7 // remaining blocks after groups
for range repeat {
s := src[base:][0:128]
d := dst[base:][0:128]
@@ -487,7 +502,7 @@ func encrypt16(block cipher.Block, dst, src, buf []byte) {
}
}
// decryption
// decrypt dispatches to the appropriate block-size-specific CFB decryptor.
func decrypt(block cipher.Block, dst, src, buf []byte) {
switch block.BlockSize() {
case 8:
@@ -499,17 +514,19 @@ func decrypt(block cipher.Block, dst, src, buf []byte) {
}
}
// decrypt 8 bytes block
// decrypt8 performs CFB decryption for 8-byte block ciphers.
// Uses double-buffering (tbl/next) with 8x loop unrolling
// to break the data dependency chain between consecutive blocks.
func decrypt8(block cipher.Block, dst, src, buf []byte) {
tbl := buf[0:8]
next := buf[8:16]
block.Encrypt(tbl, initialVector)
n := len(src) >> 3 // len(src) / 8
n := len(src) >> 3 // number of full 8-byte blocks
base := 0
repeat := n >> 3 // n / 8
left := n & 7 // n % 8
repeat := n >> 3 // number of 8-block groups (64 bytes each)
left := n & 7 // remaining blocks after groups
// loop unrolling to relieve data dependency
// 8x loop unrolling: alternates tbl/next to relieve data dependency
for range repeat {
s := src[base:][0:64]
d := dst[base:][0:64]
@@ -588,16 +605,18 @@ func decrypt8(block cipher.Block, dst, src, buf []byte) {
}
}
// decrypt16 performs CFB decryption for 16-byte block ciphers.
// Uses double-buffering (tbl/next) with 8x loop unrolling.
func decrypt16(block cipher.Block, dst, src, buf []byte) {
tbl := buf[0:16]
next := buf[16:32]
block.Encrypt(tbl, initialVector)
n := len(src) >> 4 // len(src) / 16
n := len(src) >> 4 // number of full 16-byte blocks
base := 0
repeat := n >> 3 // n / 8
left := n & 7 // n % 8
repeat := n >> 3 // number of 8-block groups (128 bytes each)
left := n & 7 // remaining blocks after groups
// loop unrolling to relieve data dependency
// 8x loop unrolling: alternates tbl/next to relieve data dependency
for range repeat {
s := src[base:][0:128]
d := dst[base:][0:128]
fec.go
+17 -12
View File
@@ -48,13 +48,14 @@ import (
"github.com/klauspost/reedsolomon"
)
// FEC (Forward Error Correction) protocol constants
const (
fecHeaderSize = 6
fecHeaderSizePlus2 = fecHeaderSize + 2 // plus 2B data size
typeData = 0xf1
typeParity = 0xf2
typeOOB = 0xf3
maxShardSets = 3
fecHeaderSize = 6 // FEC header: seqid(4B) + type(2B)
fecHeaderSizePlus2 = fecHeaderSize + 2 // FEC header + 2B payload size field
typeData = 0xf1 // FEC packet type: data shard
typeParity = 0xf2 // FEC packet type: parity shard
typeOOB = 0xf3 // FEC packet type: out-of-band (unreliable)
maxShardSets = 3 // max concurrent shard sets before discarding old ones
)
// fecPacket is a decoded FEC packet
@@ -104,7 +105,9 @@ func (h *shardHeap) Has(sn uint32) bool {
return exists
}
// fecDecoder for decoding incoming packets
// fecDecoder for decoding incoming packets.
// It collects shards grouped by shard ID and attempts Reed-Solomon recovery
// when enough shards have been received.
type fecDecoder struct {
dataShards int
parityShards int
@@ -123,9 +126,10 @@ type fecDecoder struct {
// RS decoder
codec reedsolomon.Encoder
// auto tune fec parameter
// auto-tuning: dynamically adjusts dataShards/parityShards ratio
// by detecting the period of data vs parity pulses in the incoming stream
autoTune autoTune
shouldTune bool
shouldTune bool // true when a type mismatch is detected, triggering auto-tune
}
func newFECDecoder(dataShards, parityShards int) *fecDecoder {
@@ -178,10 +182,11 @@ func (dec *fecDecoder) decode(in fecPacket) (recovered [][]byte) {
}
}
// perform auto-tuning if the decoder is out of sync.
// perform auto-tuning if the decoder detects packet type mismatches.
// This means the peer has changed FEC parameters and we must adapt.
if dec.shouldTune {
autoDS := dec.autoTune.FindPeriod(true)
autoPS := dec.autoTune.FindPeriod(false)
autoDS := dec.autoTune.FindPeriod(true) // detect data shard period
autoPS := dec.autoTune.FindPeriod(false) // detect parity shard period
// validate the auto-tuned parameters
if autoDS > 0 && autoPS > 0 && autoDS+autoPS < 256 {
kcp.go
+119 -63
View File
@@ -29,29 +29,39 @@ import (
"time"
)
// KCP Protocol Constants
const (
IKCP_RTO_NDL = 30 // no delay min rto
IKCP_RTO_MIN = 100 // normal min rto
IKCP_RTO_DEF = 200
IKCP_RTO_MAX = 60000
IKCP_CMD_PUSH = 81 // cmd: push data
IKCP_CMD_ACK = 82 // cmd: ack
IKCP_CMD_WASK = 83 // cmd: window probe (ask)
IKCP_CMD_WINS = 84 // cmd: window size (tell)
IKCP_ASK_SEND = 1 // need to send IKCP_CMD_WASK
IKCP_ASK_TELL = 2 // need to send IKCP_CMD_WINS
IKCP_WND_SND = 32
IKCP_WND_RCV = 32
IKCP_MTU_DEF = 1400
IKCP_ACK_FAST = 3
IKCP_INTERVAL = 100
IKCP_OVERHEAD = 24
IKCP_DEADLINK = 20
IKCP_THRESH_INIT = 2
IKCP_THRESH_MIN = 2
IKCP_PROBE_INIT = 500 // 500ms to probe window size
IKCP_PROBE_LIMIT = 120000 // up to 120 secs to probe window
IKCP_SN_OFFSET = 12
// Retransmission Timeout (RTO) bounds, in milliseconds
IKCP_RTO_NDL = 30 // no-delay mode: minimum RTO (ms)
IKCP_RTO_MIN = 100 // normal mode: minimum RTO (ms)
IKCP_RTO_DEF = 200 // default RTO (ms)
IKCP_RTO_MAX = 60000 // maximum RTO (ms), 60 seconds
// Command types for the KCP segment header (cmd field)
IKCP_CMD_PUSH = 81 // cmd: push data
IKCP_CMD_ACK = 82 // cmd: acknowledge
IKCP_CMD_WASK = 83 // cmd: window probe request (ask)
IKCP_CMD_WINS = 84 // cmd: window size response (tell)
// Probe flags (bitfield), set in kcp.probe to schedule probe commands
IKCP_ASK_SEND = 1 // schedule sending IKCP_CMD_WASK
IKCP_ASK_TELL = 2 // schedule sending IKCP_CMD_WINS
// Default window and MTU sizes
IKCP_WND_SND = 32 // default send window size (packets)
IKCP_WND_RCV = 32 // default receive window size (packets)
IKCP_MTU_DEF = 1400 // default MTU (bytes, not including UDP/IP header)
// Protocol parameters
IKCP_ACK_FAST = 3 // fast retransmit trigger threshold (duplicate ACK count)
IKCP_INTERVAL = 100 // default flush interval (ms)
IKCP_OVERHEAD = 24 // per-segment header size: conv(4) + cmd(1) + frg(1) + wnd(2) + ts(4) + sn(4) + una(4) + len(4)
IKCP_DEADLINK = 20 // max retransmissions before declaring dead link
IKCP_THRESH_INIT = 2 // initial slow-start threshold (packets)
IKCP_THRESH_MIN = 2 // minimum slow-start threshold (packets)
IKCP_PROBE_INIT = 500 // initial window probe timeout (ms)
IKCP_PROBE_LIMIT = 120000 // maximum window probe timeout (ms), 120 seconds
IKCP_SN_OFFSET = 12 // byte offset of sequence number (sn) within the segment header
)
type PacketType int8
@@ -179,35 +189,62 @@ func (h *segmentHeap) Has(sn uint32) bool {
return exists
}
// KCP defines a single KCP connection
// KCP defines a single KCP connection's protocol state machine.
// It is a pure ARQ (Automatic Repeat reQuest) implementation with no I/O.
type KCP struct {
conv, mtu, mss, state uint32
snd_una, snd_nxt, rcv_nxt uint32
ssthresh uint32
rx_rttvar, rx_srtt int32
rx_rto, rx_minrto uint32
snd_wnd, rcv_wnd, rmt_wnd, cwnd, probe uint32
interval, ts_flush uint32
nodelay, updated uint32
ts_probe, probe_wait uint32
dead_link, incr uint32
// Connection identity and framing
conv uint32 // conversation id, must be equal on both sides
mtu uint32 // maximum transmission unit (bytes)
mss uint32 // maximum segment size = mtu - IKCP_OVERHEAD
state uint32 // connection state, 0 = active, 0xFFFFFFFF = dead link
fastresend int32
nocwnd, stream int32
// Sequence numbers and acknowledgment tracking
snd_una uint32 // oldest unacknowledged sequence number
snd_nxt uint32 // next sequence number to send
rcv_nxt uint32 // next expected sequence number to receive
// Congestion control (RFC 5681 / RFC 6937)
ssthresh uint32 // slow-start threshold (packets)
rx_rttvar, rx_srtt int32 // RTT variance and smoothed RTT (ms), per RFC 6298
rx_rto, rx_minrto uint32 // retransmission timeout and its lower bound (ms)
snd_wnd uint32 // local send window size (packets)
rcv_wnd uint32 // local receive window size (packets)
rmt_wnd uint32 // remote advertised window size (packets)
cwnd uint32 // congestion window (packets)
incr uint32 // bytes accumulated for cwnd increment
// Window probing
probe uint32 // probe flags (IKCP_ASK_SEND / IKCP_ASK_TELL)
ts_probe uint32 // timestamp for next window probe (ms)
probe_wait uint32 // current probe timeout (ms), doubles on each retry
// Timers and scheduling
interval uint32 // flush interval (ms)
ts_flush uint32 // next flush timestamp (ms)
nodelay uint32 // 0: normal, 1: no-delay mode (reduces RTO aggressively)
updated uint32 // whether Update() has been called at least once
// Reliability
dead_link uint32 // max retransmit count before link is considered dead
fastresend int32 // fast retransmit trigger count, 0 = disabled
nocwnd int32 // 1 = disable congestion control
stream int32 // 1 = stream mode (no message boundaries), 0 = message mode
// Logging
logmask KCPLogType
snd_queue *RingBuffer[segment]
rcv_queue *RingBuffer[segment]
snd_buf *RingBuffer[segment]
rcv_buf *segmentHeap
// Data queues and buffers
snd_queue *RingBuffer[segment] // send queue: segments waiting to enter the send window
rcv_queue *RingBuffer[segment] // receive queue: ordered segments ready for user read
snd_buf *RingBuffer[segment] // send buffer: segments in-flight (sent but unacknowledged)
rcv_buf *segmentHeap // receive buffer: out-of-order segments awaiting reordering
acklist []ackItem
acklist []ackItem // pending ACKs to be flushed
buffer []byte
output output_callback
buffer []byte // pre-allocated encoding buffer for flush()
output output_callback // callback to write data to the underlying transport
log logoutput_callback
log logoutput_callback // trace log callback
}
type ackItem struct {
@@ -406,8 +443,9 @@ func (kcp *KCP) Send(buffer []byte) int {
return 0
}
// update_ack updates the smoothed RTT and RTO based on a new RTT sample.
// Algorithm follows RFC 6298: Computing TCP's Retransmission Timer.
func (kcp *KCP) update_ack(rtt int32) {
// https://tools.ietf.org/html/rfc6298
var rto uint32
if kcp.rx_srtt == 0 {
kcp.rx_srtt = rtt
@@ -431,6 +469,7 @@ func (kcp *KCP) update_ack(rtt int32) {
kcp.rx_rto = min(max(kcp.rx_minrto, rto), IKCP_RTO_MAX)
}
// shrink_buf advances snd_una to the oldest unacknowledged segment in snd_buf.
func (kcp *KCP) shrink_buf() {
if seg, ok := kcp.snd_buf.Peek(); ok {
kcp.snd_una = seg.sn
@@ -439,6 +478,9 @@ func (kcp *KCP) shrink_buf() {
}
}
// parse_ack marks a segment as acknowledged in snd_buf by sequence number.
// The segment is not removed immediately; it stays until snd_una advances past it,
// avoiding expensive shifts in the ring buffer.
func (kcp *KCP) parse_ack(sn uint32) {
if _itimediff(sn, kcp.snd_una) < 0 || _itimediff(sn, kcp.snd_nxt) >= 0 {
return
@@ -460,6 +502,8 @@ func (kcp *KCP) parse_ack(sn uint32) {
}
}
// parse_fastack increments the fast-ack counter for segments with sn < the given sn.
// Returns 1 if any segment's fastack counter has reached the fast retransmit threshold.
func (kcp *KCP) parse_fastack(sn, ts uint32) int {
shouldFastAck := 0
if _itimediff(sn, kcp.snd_una) < 0 || _itimediff(sn, kcp.snd_nxt) >= 0 {
@@ -482,6 +526,8 @@ func (kcp *KCP) parse_fastack(sn, ts uint32) int {
return shouldFastAck
}
// parse_una removes all segments from snd_buf that have been cumulatively acknowledged
// (i.e., segments with sn < una). Returns the number of segments removed.
func (kcp *KCP) parse_una(una uint32) int {
count := 0
for seg := range kcp.snd_buf.ForEach {
@@ -643,7 +689,8 @@ func (kcp *KCP) Input(data []byte, pktType PacketType, ackNoDelay bool) int {
}
}
// cwnd update when packet arrived
// Congestion window (cwnd) update on ACK arrival.
// Uses Reno-style algorithm: slow-start below ssthresh, then AIMD.
if kcp.nocwnd == 0 {
if _itimediff(kcp.snd_una, snd_una) > 0 {
if kcp.cwnd < kcp.rmt_wnd {
@@ -698,7 +745,17 @@ func (kcp *KCP) wnd_unused() uint16 {
return 0
}
// flush pending data
// flush sends pending data through the KCP output callback.
// This is the core scheduling function, organized in 6 phases:
//
// Phase 1: Flush pending ACKs
// Phase 2: Window probing (when remote window is zero)
// Phase 3: Send window probe commands (WASK/WINS)
// Phase 4: Move segments from snd_queue to snd_buf (sliding window)
// Phase 5: Retransmit segments (initial, fast, early, RTO)
// Phase 6: Update SNMP counters and congestion window
//
// Returns the suggested interval (ms) until the next flush call.
func (kcp *KCP) flush(flushType FlushType) (nextUpdate uint32) {
var seg segment
seg.conv = kcp.conv
@@ -733,9 +790,7 @@ func (kcp *KCP) flush(flushType FlushType) (nextUpdate uint32) {
atomic.StoreUint64(&DefaultSnmp.RingBufferSndBuffer, uint64(kcp.snd_buf.Len()))
}()
/*
* flush acknowledges
*/
// --- Phase 1: Flush pending ACKs ---
if flushType == IKCP_FLUSH_ACKONLY || flushType == IKCP_FLUSH_FULL {
for i, ack := range kcp.acklist {
makeSpace(IKCP_OVERHEAD)
@@ -749,7 +804,7 @@ func (kcp *KCP) flush(flushType FlushType) (nextUpdate uint32) {
kcp.acklist = kcp.acklist[0:0]
}
// probe window size (if remote window size equals zero)
// --- Phase 2: Window probing (when remote window is zero) ---
if kcp.rmt_wnd == 0 {
current := currentMs()
if kcp.probe_wait == 0 {
@@ -773,9 +828,7 @@ func (kcp *KCP) flush(flushType FlushType) (nextUpdate uint32) {
kcp.probe_wait = 0
}
/*
* flush window probing commands
*/
// --- Phase 3: Flush window probing commands ---
if (kcp.probe & IKCP_ASK_SEND) != 0 {
seg.cmd = IKCP_CMD_WASK
makeSpace(IKCP_OVERHEAD)
@@ -793,13 +846,13 @@ func (kcp *KCP) flush(flushType FlushType) (nextUpdate uint32) {
kcp.probe = 0
// calculate window size
// --- Phase 4: Move segments from snd_queue to snd_buf (sliding window) ---
// Effective window = min(snd_wnd, rmt_wnd, cwnd)
cwnd := min(kcp.snd_wnd, kcp.rmt_wnd)
if kcp.nocwnd == 0 {
cwnd = min(kcp.cwnd, cwnd)
}
// sliding window, controlled by snd_nxt && sna_una+cwnd
newSegsCount := 0
for {
if _itimediff(kcp.snd_nxt, kcp.snd_una+cwnd) >= 0 {
@@ -825,9 +878,12 @@ func (kcp *KCP) flush(flushType FlushType) (nextUpdate uint32) {
resent = 0xffffffff
}
/*
* flush segments
*/
// --- Phase 5: Retransmit segments from snd_buf ---
// Determines which segments need (re)transmission:
// - Initial transmit (xmit == 0)
// - Fast retransmit (fastack >= fastresend threshold)
// - Early retransmit (fastack > 0, no new segments queued)
// - RTO-based retransmit (current >= resendts)
current := currentMs()
var change, lostSegs, fastRetransSegs, earlyRetransSegs uint64
nextUpdate = kcp.interval
@@ -884,7 +940,7 @@ func (kcp *KCP) flush(flushType FlushType) (nextUpdate uint32) {
kcp.debugLog(IKCP_LOG_OUT_PUSH, "conv", segment.conv, "sn", segment.sn, "frg", segment.frg, "una", segment.una, "ts", segment.ts, "xmit", segment.xmit, "datalen", len(segment.data))
if segment.xmit >= kcp.dead_link {
kcp.state = 0xFFFFFFFF
kcp.state = 0xFFFFFFFF // mark connection as dead
}
}
@@ -895,7 +951,7 @@ func (kcp *KCP) flush(flushType FlushType) (nextUpdate uint32) {
}
}
// counter updates
// --- Phase 6: Update SNMP counters and congestion window ---
sum := lostSegs
if lostSegs > 0 {
atomic.AddUint64(&DefaultSnmp.LostSegs, lostSegs)
@@ -914,8 +970,8 @@ func (kcp *KCP) flush(flushType FlushType) (nextUpdate uint32) {
// cwnd update
if kcp.nocwnd == 0 {
// update ssthresh
// rate halving, https://tools.ietf.org/html/rfc6937
// Update ssthresh after fast retransmit.
// Rate halving per RFC 6937: ssthresh = inflight / 2
if change > 0 {
inflight := kcp.snd_nxt - kcp.snd_una
kcp.ssthresh = max(inflight/2, IKCP_THRESH_MIN)
@@ -923,7 +979,7 @@ func (kcp *KCP) flush(flushType FlushType) (nextUpdate uint32) {
kcp.incr = kcp.cwnd * kcp.mss
}
// congestion control, https://tools.ietf.org/html/rfc5681
// Congestion control after RTO: reset cwnd per RFC 5681
if lostSegs > 0 {
kcp.ssthresh = max(cwnd/2, IKCP_THRESH_MIN)
kcp.cwnd = 1
ringbuffer.go
+2 -2
View File
@@ -23,8 +23,8 @@
package kcp
const (
RINGBUFFER_MIN = 8
RINGBUFFER_EXP = 1024
RINGBUFFER_MIN = 8 // minimum ring buffer capacity
RINGBUFFER_EXP = 1024 // growth threshold: below this, double; above this, grow by 25%
)
// RingBuffer is a generic ring (circular) buffer that supports dynamic resizing.
sess.go
+52 -18
View File
@@ -62,32 +62,35 @@ import (
"golang.org/x/time/rate"
)
// Session-layer constants
const (
// 16-bytes nonce for each packet
nonceSize = 16
// 4-bytes packet checksum
// 4-bytes CRC32 checksum per packet
crcSize = 4
// overall crypto header size
// overall crypto header size: nonce + CRC32
cryptHeaderSize = nonceSize + crcSize
// maximum packet size
// maximum packet size (Ethernet MTU)
mtuLimit = 1500
// conv field size
// conversation ID field size (bytes)
convSize = 4
// accept backlog
// accept backlog: max pending connections for Listener
acceptBacklog = 128
// dev backlog
// devBacklog: channel buffer size for post-processing pipeline
devBacklog = 2048
// max latency for consecutive FEC encoding, in millisecond
// max latency for consecutive FEC encoding (ms).
// If the interval between two data packets exceeds this,
// parity generation is skipped.
maxFECEncodeLatency = 500
// max batch size
// max number of packets batched in a single sendmmsg/writev call
maxBatchSize = 64
)
@@ -638,10 +641,13 @@ func (s *UDPSession) Control(f func(conn net.PacketConn) error) error {
return f(s.conn)
}
// a goroutine to handle post processing of kcp and make the critical section smaller
// pipeline for outgoing packets (from ARQ to network)
// postProcess is the goroutine that handles the outgoing packet pipeline.
// It runs the following stages sequentially for each packet:
// 1. FEC encoding — generate parity shards (Reed-Solomon)
// 2. Encryption — AEAD (e.g. AES-GCM) or CFB mode with CRC32
// 3. TX batching — accumulate packets and flush via sendmmsg/writev
//
// KCP output -> FEC encoding -> CRC32 integrity -> Encryption -> TxQueue
// Pipeline: KCP output -> chPostProcessing -> [FEC] -> [Encrypt] -> TxQueue -> Network
func (s *UDPSession) postProcess() {
txqueue := make([]ipv4.Message, 0, devBacklog)
chDie := s.die
@@ -656,7 +662,7 @@ func (s *UDPSession) postProcess() {
var ecc [][]byte
// 1. FEC encoding
// --- Stage 1: FEC encoding ---
if s.fecEncoder != nil {
if !oob {
ecc = s.fecEncoder.encode(buf, maxFECEncodeLatency)
@@ -665,7 +671,10 @@ func (s *UDPSession) postProcess() {
}
}
// 2. Encryption
// --- Stage 2: Encryption ---
// Two modes supported:
// - AEAD (e.g. AES-GCM): nonce + authenticated ciphertext, no separate CRC
// - CFB (legacy block ciphers): random nonce + CRC32 checksum + CFB encryption
switch block := s.block.(type) {
case nil:
case *aeadCrypt: // AEAD mode
@@ -700,7 +709,7 @@ func (s *UDPSession) postProcess() {
}
}
// 3. TxQueue
// --- Stage 3: TX batching ---
var msg ipv4.Message
msg.Addr = s.remote
@@ -920,8 +929,14 @@ func (s *UDPSession) notifyWriteError(err error) {
})
}
// packet input pipeline:
// network -> [decryption ->] [crc32 ->] [FEC ->] [KCP input ->] stream -> application
// -----------------------------------------------------------------------
// Packet input pipeline (decryption -> integrity check -> FEC -> KCP)
// -----------------------------------------------------------------------
// packetInput is the entry point for incoming packets.
// It handles decryption and CRC32 verification before passing data to kcpInput.
//
// Pipeline: Network -> [Decrypt] -> [CRC32] -> kcpInput
func (s *UDPSession) packetInput(data []byte) {
switch block := s.block.(type) {
case nil:
@@ -969,7 +984,16 @@ func (s *UDPSession) packetInput(data []byte) {
s.kcpInput(data)
}
// kcpInput inputs a decrypted and crc32-checked packet into kcp with FEC handling
// kcpInput routes a decrypted packet into the KCP state machine,
// handling FEC decoding and OOB delivery.
//
// Packet demultiplexing uses the 16-bit field at offset 4:
// - 0xf1 (typeData) / 0xf2 (typeParity): FEC-encoded packet
// - 0xf3 (typeOOB): out-of-band packet (unreliable, bypasses KCP)
// - other values: raw KCP packet (no FEC)
//
// Note: KCP cmd values [81-84] with frg [0-255] do not collide with
// FEC type markers 0x00f1/0x00f2/0x00f3 in little-endian.
func (s *UDPSession) kcpInput(data []byte) {
atomic.AddUint64(&DefaultSnmp.InPkts, 1)
atomic.AddUint64(&DefaultSnmp.InBytes, uint64(len(data)))
@@ -1066,6 +1090,10 @@ func (s *UDPSession) kcpInput(data []byte) {
}
}
// -----------------------------------------------------------------------
// Listener: server-side session multiplexer
// -----------------------------------------------------------------------
type (
// Listener defines a server which will be waiting to accept incoming connections
Listener struct {
@@ -1091,7 +1119,9 @@ type (
}
)
// packet input stage
// packetInput is the Listener's packet input handler.
// It decrypts the packet, demultiplexes by remote address,
// and dispatches to existing sessions or creates new ones.
func (l *Listener) packetInput(data []byte, addr net.Addr) {
switch block := l.block.(type) {
case nil:
@@ -1363,6 +1393,10 @@ func (l *Listener) Addr() net.Addr {
return l.conn.LocalAddr()
}
// -----------------------------------------------------------------------
// Public API: Dial, Listen, and connection factory functions
// -----------------------------------------------------------------------
// Listen listens for incoming KCP packets addressed to the local address laddr on the network "udp",
func Listen(laddr string) (net.Listener, error) {
return ListenWithOptions(laddr, nil, 0, 0)
snmp.go
+3 -3
View File
@@ -44,9 +44,9 @@ type Snmp struct {
OutSegs uint64 // outgoing KCP segments
InBytes uint64 // UDP bytes received
OutBytes uint64 // UDP bytes sent
RetransSegs uint64 // accmulated retransmited segments
FastRetransSegs uint64 // accmulated fast retransmitted segments
EarlyRetransSegs uint64 // accmulated early retransmitted segments
RetransSegs uint64 // accumulated retransmitted segments
FastRetransSegs uint64 // accumulated fast retransmitted segments
EarlyRetransSegs uint64 // accumulated early retransmitted segments
LostSegs uint64 // number of segs inferred as lost
RepeatSegs uint64 // number of segs duplicated
FECFullShardSet uint64 // number of FEC segments that are full
timedsched.go
+26 -6
View File
@@ -29,7 +29,8 @@ import (
"time"
)
// SystemTimedSched is the library level timed-scheduler
// SystemTimedSched is the library-level timed scheduler, shared by all sessions.
// It drives periodic KCP flush()/update() calls, avoiding one goroutine per session.
var SystemTimedSched *TimedSched = NewTimedSched(runtime.NumCPU())
type timedFunc struct {
@@ -53,14 +54,31 @@ func (h *timedFuncHeap) Pop() any {
return x
}
// TimedSched represents the control struct for timed parallel scheduler
// TimedSched is a two-stage parallel scheduler for timed task execution.
//
// Architecture (two-stage pipeline):
//
// Stage 1 - "prepend" goroutine:
// External callers submit tasks via Put(). Tasks are appended to a shared
// slice under a mutex (fast, non-blocking). The prepend goroutine drains
// this slice and feeds tasks one-by-one into chTask.
//
// Stage 2 - "sched" goroutines (N = NumCPU):
// Each sched goroutine maintains a local min-heap of pending tasks.
// It receives tasks from chTask, executes overdue ones immediately,
// and uses a timer for the earliest future task.
//
// Why two stages?
// - Stage 1 decouples callers from the scheduler's internal heap,
// ensuring Put() never blocks on heap operations.
// - Stage 2 runs in parallel, distributing timer-driven work across CPUs.
type TimedSched struct {
// prepending tasks
// Stage 1: task collection
prependTasks []timedFunc
prependLock sync.Mutex
chPrependNotify chan struct{}
// tasks will be distributed through chTask
// Stage 2: parallel execution
chTask chan timedFunc
dieOnce sync.Once
@@ -81,7 +99,8 @@ func NewTimedSched(parallel int) *TimedSched {
return ts
}
// sched is a goroutine to schedule and execute timed tasks.
// sched is a worker goroutine (Stage 2) that manages a local min-heap
// of timed tasks. It executes tasks when their deadline arrives.
func (ts *TimedSched) sched() {
timer := time.NewTimer(0)
defer timer.Stop()
@@ -122,7 +141,8 @@ func (ts *TimedSched) sched() {
}
}
// prepend is the front desk goroutine to register tasks
// prepend is the Stage 1 goroutine that collects externally submitted tasks
// and feeds them into the Stage 2 worker pool via chTask.
func (ts *TimedSched) prepend() {
var tasks []timedFunc
for {