Documentation
¶
Overview ¶
Package tsn provides Time-Sensitive Networking stream descriptors and configuration utilities following the OMG DDS-TSN specification.
A TSN stream maps a DDS topic to a bounded-latency IEEE 802.1 network flow. Stream descriptors control VLAN tagging, DSCP marking, frame size bounds, and scheduled transmit timing (SO_TXTIME / ETF qdisc).
Usage — load from a JSON config file:
cfg, err := tsn.LoadConfig("tsn_streams.json")
p, err := rtps.New(0, rtps.WithTSNConfig(cfg))
Config file format (JSON):
{
"streams": [
{
"topic": "vehicle/speed",
"vid": 100,
"pcp": 5,
"dscp": 46,
"max_frame_size": 1500,
"max_interval_frames": 1,
"interval_us": 125,
"tx_offset_us": 50,
"talker_id": "ecu-cluster-1"
}
]
}
Index ¶
Constants ¶
This section is empty.
Variables ¶
var ErrNoStream = errors.New("tsn: no stream configured for topic")
ErrNoStream is returned when StreamConfig.StreamForTopic finds no match.
var ErrNotSupported = errors.New("tsn: TAPRIO qdisc requires Linux")
ErrNotSupported is returned by TAPRIOConfig.Apply on platforms other than Linux.
Functions ¶
This section is empty.
Types ¶
type HealthTracker ¶ added in v0.7.0
type HealthTracker struct {
// contains filtered or unexported fields
}
HealthTracker monitors actual write timestamps against a stream's scheduled interval and computes timing health statistics. It is safe for concurrent use.
func NewHealthTracker ¶ added in v0.7.0
func NewHealthTracker(s *Stream, windowSize int) *HealthTracker
NewHealthTracker creates a HealthTracker for stream. windowSize is the maximum number of past writes kept for health calculation; values ≤ 0 are clamped to 100.
func (*HealthTracker) Health ¶ added in v0.7.0
func (h *HealthTracker) Health() StreamHealth
Health returns a StreamHealth snapshot computed from the recorded writes.
func (*HealthTracker) Record ¶ added in v0.7.0
func (h *HealthTracker) Record(actual time.Time)
Record registers that a write occurred at actual. It computes the nearest scheduled interval boundary and stores the lateness for health tracking.
func (*HealthTracker) Reset ¶ added in v0.7.0
func (h *HealthTracker) Reset()
Reset clears all recorded write history.
type Stream ¶
type Stream struct {
// Topic is the DDS topic name this stream applies to (exact match).
Topic string `json:"topic"`
// VID is the IEEE 802.1Q VLAN ID (0 = untagged).
VID uint16 `json:"vid"`
// PCP is the VLAN Priority Code Point (0–7). Linux maps PCP to a
// traffic class via tc; the kernel also exposes it as SO_PRIORITY.
PCP uint8 `json:"pcp"`
// DSCP is the IP Differentiated Services Code Point (0–63).
// Written into the IP ToS field (DSCP << 2) via IP_TOS.
DSCP uint8 `json:"dscp"`
// MaxFrameSize is the maximum Ethernet payload bytes per frame.
// Use 1500 for standard Ethernet (no jumbo frames).
// The RTPS layer uses this to bound DATA_FRAG payload size.
MaxFrameSize int `json:"max_frame_size"`
// MaxIntervalFrames is the maximum number of frames allowed per Interval.
MaxIntervalFrames int `json:"max_interval_frames"`
// IntervalUS is the TSN transmit interval in microseconds.
// 125 µs = 8000 fps (Audio Video Bridging Class A / AVTP).
IntervalUS int64 `json:"interval_us"`
// TxOffsetUS is the transmit offset within the Interval in microseconds.
// Used with SO_TXTIME / ETF qdisc for scheduled transmit.
TxOffsetUS int64 `json:"tx_offset_us"`
// TalkerID is an informational talker identifier used when programming
// VLAN / CNC (e.g., IEEE 802.1Qcc YANG model TalkerID).
TalkerID string `json:"talker_id"`
}
Stream is a DDS-TSN stream descriptor as defined in the OMG DDS-TSN specification (OMG Document ptc/2021-03-01). It binds a DDS topic to a TSN flow with specific IEEE 802.1 network scheduling constraints.
func (*Stream) MaxFragPayload ¶
MaxFragPayload returns the maximum RTPS fragment payload bytes for this stream, reserving space for RTPS headers (~48 bytes). Returns 0 when MaxFrameSize is unset (meaning no fragmentation bound is configured).
type StreamConfig ¶
type StreamConfig struct {
Streams []Stream `json:"streams"`
}
StreamConfig is the top-level structure for a tsn_streams JSON file.
func LoadConfig ¶
func LoadConfig(path string) (*StreamConfig, error)
LoadConfig reads and parses a TSN stream configuration from a JSON file. The file must contain a top-level "streams" array of Stream objects.
func ParseConfig ¶
func ParseConfig(data []byte) (*StreamConfig, error)
ParseConfig parses TSN stream configuration from JSON bytes.
func (*StreamConfig) StreamForTopic ¶
func (c *StreamConfig) StreamForTopic(topic string) *Stream
StreamForTopic returns a pointer to the first Stream whose Topic matches, or nil if no stream is configured for the given topic.
func (*StreamConfig) Topics ¶
func (c *StreamConfig) Topics() []string
Topics returns the set of all configured topic names.
type StreamHealth ¶ added in v0.7.0
type StreamHealth struct {
Topic string
Interval time.Duration
TxOffset time.Duration
WriteCount uint64
LateWrites uint64
MaxLateness time.Duration
Healthy bool
}
StreamHealth is a point-in-time timing health snapshot for a TSN stream. Healthy is true when fewer than 5 % of writes in the observation window were late (lateness > 20 µs past the nearest scheduled slot).
type TAPRIOConfig ¶ added in v0.7.0
type TAPRIOConfig struct {
// CycleTime is the total schedule cycle (sum of all stream intervals).
CycleTime time.Duration
// Entries is the ordered list of gate control entries.
Entries []TAPRIOEntry
// Interface is the network interface to configure (e.g., "eth0").
// Required when calling Apply.
Interface string
// BaseTime is the TAPRIO schedule base time in CLOCK_TAI nanoseconds.
// Zero means the kernel picks the next cycle boundary.
BaseTime int64
// Offload, when true, requests full hardware offload from the NIC.
Offload bool
}
TAPRIOConfig holds a TAPRIO gate control list derived from a StreamConfig. Use TAPRIOFromStreams to build one; call TCCommand to get a tc(8) command string, or call Apply (Linux only) to program the qdisc directly via netlink.
func TAPRIOFromStreams ¶ added in v0.7.0
func TAPRIOFromStreams(cfg *StreamConfig) (*TAPRIOConfig, error)
TAPRIOFromStreams derives a simple TAPRIO gate schedule from cfg. Each stream's PCP value determines its traffic class (TC = PCP). The gate opens each TC exclusively for its TransmitInterval; other TCs are closed during that slot. Returns an error if cfg has no streams or all streams have a zero interval.
func (*TAPRIOConfig) Apply ¶ added in v0.12.1
func (c *TAPRIOConfig) Apply() error
Apply programs the TAPRIO qdisc on c.Interface by sending an RTM_NEWQDISC netlink message to the kernel. It replaces any existing root qdisc.
Requires CAP_NET_ADMIN. Returns ErrNotSupported on non-Linux platforms (see taprio_stub.go).
func (*TAPRIOConfig) CycleDuration ¶ added in v0.12.1
func (c *TAPRIOConfig) CycleDuration() time.Duration
CycleDuration returns the effective cycle time: TAPRIOConfig.CycleTime if set, otherwise the sum of all entry intervals.
func (*TAPRIOConfig) TCCommand ¶ added in v0.7.0
func (t *TAPRIOConfig) TCCommand(iface string, baseTimeNS int64) string
TCCommand returns a tc(8) command string that programs this TAPRIO schedule on iface. baseTimeNS is the TAI base time in nanoseconds (use 0 to start at the beginning of the epoch). The output is a starting-point template; operators may need to adjust the num_tc / map / queues arguments to match their specific NIC and traffic-class configuration.
func (*TAPRIOConfig) Validate ¶ added in v0.12.1
func (c *TAPRIOConfig) Validate() error
Validate checks that c is well-formed for a call to Apply.
func (*TAPRIOConfig) VerifyApplied ¶ added in v0.13.0
func (c *TAPRIOConfig) VerifyApplied() error
VerifyApplied queries the kernel via RTM_GETQDISC to confirm that a TAPRIO qdisc is the root qdisc on c.Interface. Returns nil if "taprio" is found, or an error describing the mismatch. Call this after Apply to confirm the schedule was accepted by the kernel.
type TAPRIOEntry ¶ added in v0.7.0
TAPRIOEntry is one gate control entry in an IEEE 802.1Qbv gate control list. GateMask is a bitmask of open traffic classes (bit N = TC N open).
Source Files