flowcontrol

README

Flow Control Handler

Flow control prevents buffer overflow by managing send/receive windows between client and server. It ensures the sender doesn't overwhelm the receiver with too much data.

What It Does

• Prevents Buffer Overflow: Controls data rate to match receiver's capacity
• Connection-Level: One flow controller per connection (IP:Port)
• Threshold-Based: Automatically sends window updates when 75% of receive buffer is consumed
• Auto-Tuning: Window size adapts based on network conditions and RTT

Key Features

• Symmetric design: Both client and server use the same flow control logic
• Lightweight: 9-byte fixed-size feedback packets
• Independent: Works with or without reliable transport and congestion control

Usage

Client-Side Setup

import (
    "github.com/appnet-org/arpc/pkg/custom/flowcontrol"
    "github.com/appnet-org/arpc/pkg/transport"
    "github.com/appnet-org/arpc/pkg/packet"
)

// Create UDP transport
udpTransport, _ := transport.NewUDPTransport(":0")
defer udpTransport.Close()

// Register FCFeedback packet type
fcFeedbackType, _ := udpTransport.RegisterPacketType(
    flowcontrol.FCFeedbackPacketName, 
    &flowcontrol.FCFeedbackCodec{},
)

// Create FC client handler (defaults: 15MB initial, 25MB max window)
clientFCHandler := flowcontrol.NewFCClientHandler(
    udpTransport,
    udpTransport.GetTimerManager(),
)
defer clientFCHandler.Cleanup()

// Register for REQUEST packets (OnSend)
requestChain, _ := udpTransport.GetHandlerRegistry().GetHandlerChain(
    packet.PacketTypeRequest.TypeID, 
    transport.RoleClient,
)
requestChain.AddHandler(clientFCHandler)

// Register for RESPONSE packets (OnReceive)
responseChain, _ := udpTransport.GetHandlerRegistry().GetHandlerChain(
    packet.PacketTypeResponse.TypeID, 
    transport.RoleClient,
)
responseChain.AddHandler(clientFCHandler)

// Register handler chain for FCFeedback packets
fcFeedbackChain := transport.NewHandlerChain("ClientFCFeedbackChain", clientFCHandler)
udpTransport.RegisterHandlerChain(fcFeedbackType.TypeID, fcFeedbackChain, transport.RoleClient)

Server-Side Setup

// Create UDP transport
udpTransport, _ := transport.NewUDPTransport(":8080")
defer udpTransport.Close()

// Register FCFeedback packet type
fcFeedbackType, _ := udpTransport.RegisterPacketType(
    flowcontrol.FCFeedbackPacketName, 
    &flowcontrol.FCFeedbackCodec{},
)

// Create FC server handler
serverFCHandler := flowcontrol.NewFCServerHandler(
    udpTransport,
    udpTransport.GetTimerManager(),
)
defer serverFCHandler.Cleanup()

// Register for REQUEST packets (OnReceive)
requestChain, _ := udpTransport.GetHandlerRegistry().GetHandlerChain(
    packet.PacketTypeRequest.TypeID, 
    transport.RoleServer,
)
requestChain.AddHandler(serverFCHandler)

// Register for RESPONSE packets (OnSend)
responseChain, _ := udpTransport.GetHandlerRegistry().GetHandlerChain(
    packet.PacketTypeResponse.TypeID, 
    transport.RoleServer,
)
responseChain.AddHandler(serverFCHandler)

// Register handler chain for FCFeedback packets
fcFeedbackChain := transport.NewHandlerChain("ServerFCFeedbackChain", serverFCHandler)
udpTransport.RegisterHandlerChain(fcFeedbackType.TypeID, fcFeedbackChain, transport.RoleServer)

Custom Configuration

// Create handler with custom window sizes
clientFCHandler := flowcontrol.NewFCClientHandlerWithConfig(
    udpTransport,
    udpTransport.GetTimerManager(),
    10*1024*1024, // 10 MB initial receive window (default: 15 MB)
    20*1024*1024, // 20 MB max receive window (default: 25 MB)
)

How It Works

1. Sender checks window before sending data
2. Receiver tracks bytes received and consumed
3. When 75% of receive buffer is consumed, receiver sends FCFeedback with new window size
4. Sender updates its send window and can send more data
5. Automatic cleanup removes idle connections after 30 seconds

Configuration

Default Window Sizes:

• Initial receive window: 15 MB
• Max receive window: 25 MB
• Initial send window: 0 (updated by peer's first feedback)

These defaults work well for most applications but can be customized if needed.

Documentation

Index

• Constants
• type ConnectionID
  • func (c ConnectionID) Key() uint64
  • func (c ConnectionID) String() string
• type FCClientHandler
  • func NewFCClientHandler(transportSender TransportSender, timerMgr TimerScheduler) *FCClientHandler
  • func NewFCClientHandlerWithConfig(transportSender TransportSender, timerMgr TimerScheduler, ...) *FCClientHandler
  • func (h *FCClientHandler) Cleanup()
  • func (h *FCClientHandler) OnReceive(pkt any, addr *net.UDPAddr) error
  • func (h *FCClientHandler) OnSend(pkt any, addr *net.UDPAddr) error
• type FCConnectionState
• type FCFeedbackCodec
  • func (c *FCFeedbackCodec) Deserialize(data []byte) (any, error)
  • func (c *FCFeedbackCodec) Serialize(pkt any, pool *common.BufferPool) ([]byte, error)
• type FCFeedbackPacket
• type FCHandler
  • func (h *FCHandler) Cleanup(cleanupTimerKey transport.TimerKey)
  • func (h *FCHandler) GetConnectionInfo(connID ConnectionID) (sendWindow, receiveWindow protocol.ByteCount, exists bool)
• type FCServerHandler
  • func NewFCServerHandler(transportSender TransportSender, timerMgr TimerScheduler) *FCServerHandler
  • func NewFCServerHandlerWithConfig(transportSender TransportSender, timerMgr TimerScheduler, ...) *FCServerHandler
  • func (h *FCServerHandler) Cleanup()
  • func (h *FCServerHandler) OnReceive(pkt any, addr *net.UDPAddr) error
  • func (h *FCServerHandler) OnSend(pkt any, addr *net.UDPAddr) error
• type TimerScheduler
• type TransportSender

Constants

const (
	TimerKeyFCClientCleanup transport.TimerKey = 20
	TimerKeyFCServerCleanup transport.TimerKey = 21
)

Predefined timer key constants for flow control timers

const FCFeedbackPacketName = "FCFeedback"

Types

type ConnectionID

type ConnectionID struct {
	IP   [4]byte
	Port uint16
}

ConnectionID uniquely identifies a connection

func (ConnectionID) Key

func (c ConnectionID) Key() uint64

Key returns a binary uint64 representation for efficient map key usage Format: IP (4 bytes in high 48 bits) | Port (2 bytes in low 16 bits)

func (ConnectionID) String

func (c ConnectionID) String() string

String returns a string representation of the connection ID for logging

type FCClientHandler

type FCClientHandler struct {
	*FCHandler // Embed base handler
}

FCClientHandler implements the client-side flow control logic

func NewFCClientHandler

func NewFCClientHandler(
	transportSender TransportSender,
	timerMgr TimerScheduler,
) *FCClientHandler

NewFCClientHandler creates a new flow control client handler with default configuration

func NewFCClientHandlerWithConfig

func NewFCClientHandlerWithConfig(
	transportSender TransportSender,
	timerMgr TimerScheduler,
	initialReceiveWindow protocol.ByteCount,
	maxReceiveWindow protocol.ByteCount,
) *FCClientHandler

NewFCClientHandlerWithConfig creates a new flow control client handler with custom configuration

func (*FCClientHandler) Cleanup

func (h *FCClientHandler) Cleanup()

Cleanup cleans up resources

func (*FCClientHandler) OnReceive

func (h *FCClientHandler) OnReceive(pkt any, addr *net.UDPAddr) error

OnReceive handles incoming packets (client side) Tracks RESPONSE packets and processes FCFeedback packets

func (*FCClientHandler) OnSend

func (h *FCClientHandler) OnSend(pkt any, addr *net.UDPAddr) error

OnSend handles outgoing packets (client side) Tracks REQUEST packets and ignores FCFeedback packets (just updates activity)

type FCConnectionState

type FCConnectionState struct {
	ConnID         ConnectionID
	LastActivity   time.Time
	FlowController flowcontrol.ConnectionFlowController // Connection-level flow control
}

FCConnectionState tracks flow control state for a single connection

type FCFeedbackCodec

type FCFeedbackCodec struct{}

FCFeedbackCodec implements PacketCodec for FCFeedback packets

func (*FCFeedbackCodec) Deserialize

func (c *FCFeedbackCodec) Deserialize(data []byte) (any, error)

Deserialize decodes binary data into a FCFeedbackPacket

func (*FCFeedbackCodec) Serialize

func (c *FCFeedbackCodec) Serialize(pkt any, pool *common.BufferPool) ([]byte, error)

Serialize encodes a FCFeedbackPacket into binary format: [PacketTypeID(1B)][SendWindow(8B)] Total: 9 bytes fixed size

type FCFeedbackPacket

type FCFeedbackPacket struct {
	PacketTypeID packet.PacketTypeID // 1 byte
	SendWindow   uint64              // 8 bytes - new send window offset
}

FCFeedbackPacket provides flow control window updates This packet is sent when the receive window needs to be updated (threshold-based), allowing the sender to continue sending data without being flow-control blocked.

type FCHandler

type FCHandler struct {
	// contains filtered or unexported fields
}

FCHandler is the base handler containing common state and logic

func (*FCHandler) Cleanup

func (h *FCHandler) Cleanup(cleanupTimerKey transport.TimerKey)

Cleanup cleans up resources

func (*FCHandler) GetConnectionInfo

func (h *FCHandler) GetConnectionInfo(connID ConnectionID) (sendWindow, receiveWindow protocol.ByteCount, exists bool)

GetConnectionInfo returns flow control info for debugging (optional)

type FCServerHandler

type FCServerHandler struct {
	*FCHandler // Embed base handler
}

FCServerHandler implements the server-side flow control logic

func NewFCServerHandler

func NewFCServerHandler(
	transportSender TransportSender,
	timerMgr TimerScheduler,
) *FCServerHandler

NewFCServerHandler creates a new flow control server handler with default configuration

func NewFCServerHandlerWithConfig

func NewFCServerHandlerWithConfig(
	transportSender TransportSender,
	timerMgr TimerScheduler,
	initialReceiveWindow protocol.ByteCount,
	maxReceiveWindow protocol.ByteCount,
) *FCServerHandler

NewFCServerHandlerWithConfig creates a new flow control server handler with custom configuration

func (*FCServerHandler) Cleanup

func (h *FCServerHandler) Cleanup()

Cleanup cleans up resources

func (*FCServerHandler) OnReceive

func (h *FCServerHandler) OnReceive(pkt any, addr *net.UDPAddr) error

OnReceive handles incoming packets (server side) Tracks REQUEST packets and processes FCFeedback packets

func (*FCServerHandler) OnSend

func (h *FCServerHandler) OnSend(pkt any, addr *net.UDPAddr) error

OnSend handles outgoing packets (server side) Tracks RESPONSE packets and ignores FCFeedback packets (just updates activity)

type TimerScheduler

type TimerScheduler interface {
	Schedule(id transport.TimerKey, duration time.Duration, callback transport.TimerCallback)
	SchedulePeriodic(id transport.TimerKey, interval time.Duration, callback transport.TimerCallback)
	StopTimer(id transport.TimerKey) bool
}

TimerScheduler interface for managing timers

type TransportSender

type TransportSender interface {
	Send(addr string, rpcID uint64, data []byte, pktType packet.PacketType) error
	GetPacketRegistry() *packet.PacketRegistry
	GetConn() *net.UDPConn
}

TransportSender interface for sending packets (avoid circular dependency)