fix: resolve 7 code review issues from v0.4.x

- Fix KSUDP clutch field reading fuel at offset 59 (KSUDP has no clutch input)
- Replace Notify with watch::channel for reliable shutdown (no missed wakeups)
- Deduplicate read_u16 from parser_pcars into shared binary_utils
- Remove dead Display impl for FeedSource
- Remove unused read_i8 helper from binary_utils
- Change Vec<Instant> to [Instant; 50] in FeedState (eliminate heap alloc)
- Cache rarely-changing strings by driver_id in Telemetry Tool parser
This commit is contained in:
Jan-Peter von Hunnius 2026-06-04 11:12:55 +02:00
parent 8f50302a78
commit 56d3169033
7 changed files with 98 additions and 45 deletions

View file

@ -51,12 +51,6 @@ pub fn read_u16(data: &[u8], offset: usize) -> u16 {
u16::from_le_bytes(buf)
}
/// Read a signed `i8` from the data at the given offset.
/// Returns 0 if the offset is out of bounds.
pub fn read_i8(data: &[u8], offset: usize) -> i8 {
data.get(offset).map(|&b| b as i8).unwrap_or(0)
}
/// Read a little-endian `f32` from the data at the given offset.
/// Returns 0.0 if the offset is out of bounds (safe fallback for malformed packets).
pub fn read_f32(data: &[u8], offset: usize) -> f32 {

View file

@ -43,7 +43,7 @@ pub struct FeedState {
pub last_packet_ago_secs: Option<f64>,
pub start_time: Option<Instant>,
pub last_packet_time: Option<Instant>,
pub packet_timestamps: Vec<Instant>,
pub packet_timestamps: [Instant; RING_CAPACITY],
pub ring_index: usize,
pub ring_count: usize,
pub last_frame: Option<TelemetryFrame>,
@ -62,7 +62,7 @@ impl FeedState {
last_packet_ago_secs: None,
start_time: None,
last_packet_time: None,
packet_timestamps: vec![Instant::now(); RING_CAPACITY],
packet_timestamps: [Instant::now(); RING_CAPACITY],
ring_index: 0,
ring_count: 0,
last_frame: None,
@ -85,9 +85,7 @@ impl FeedState {
self.last_packet_ago_secs = Some(0.0);
if self.ring_count < RING_CAPACITY {
if self.ring_count < self.packet_timestamps.len() {
self.packet_timestamps[self.ring_count] = now;
}
self.packet_timestamps[self.ring_count] = now;
self.ring_count += 1;
self.ring_index = self.ring_count % RING_CAPACITY;
} else {

View file

@ -120,12 +120,32 @@ fn install_panic_hook() {
}));
}
struct CachedStrings {
driver_id: u32,
driver_name: Option<String>,
nationality: Option<String>,
car_name: Option<String>,
tyre_compound: Option<String>,
}
impl CachedStrings {
fn new() -> Self {
Self {
driver_id: u32::MAX,
driver_name: None,
nationality: None,
car_name: None,
tyre_compound: None,
}
}
}
async fn run_feed(
feed: FeedConfig,
socket: tokio::net::UdpSocket,
session_dir: String,
state: Arc<std::sync::Mutex<FeedState>>,
shutdown: Arc<tokio::sync::Notify>,
mut shutdown_rx: tokio::sync::watch::Receiver<bool>,
) {
let handshake_target: Option<std::net::SocketAddr> = if feed.needs_handshake {
let target: std::net::SocketAddr = match format!("{}:{}", game_host(), feed.port).parse() {
@ -268,6 +288,7 @@ async fn run_feed(
};
let mut file = BufWriter::new(file);
let mut buf = vec![0u8; BUFFER_SIZE];
let mut string_cache = CachedStrings::new();
let mut hs_interval: Option<tokio::time::Interval> = handshake_target.map(|_| {
tokio::time::interval(Duration::from_secs(HANDSHAKE_INTERVAL_SECS))
@ -295,8 +316,25 @@ async fn run_feed(
_ => None,
}
} else {
// Telemetry Tool: parse only the first driver
parser_telemetry_tool::parse_first_driver(&buf[..len]).ok()
// Telemetry Tool: parse first driver, cache rarely-changing strings
match parser_telemetry_tool::parse_first_driver_no_strings(&buf[..len]) {
Ok((mut frame, driver_id)) => {
if driver_id != string_cache.driver_id {
string_cache.driver_id = driver_id;
let raw = &buf[..len];
string_cache.driver_name = Some(binary_utils::read_string(raw, parser_telemetry_tool::STRING_OFFSET_DRIVER_NAME, parser_telemetry_tool::STRING_LEN_DRIVER_NAME));
string_cache.nationality = Some(binary_utils::read_string(raw, parser_telemetry_tool::STRING_OFFSET_NATIONALITY, parser_telemetry_tool::STRING_LEN_NATIONALITY));
string_cache.car_name = Some(binary_utils::read_string(raw, parser_telemetry_tool::STRING_OFFSET_CAR_NAME, parser_telemetry_tool::STRING_LEN_CAR_NAME));
string_cache.tyre_compound = Some(binary_utils::read_string(raw, parser_telemetry_tool::STRING_OFFSET_TYRE_COMPOUND, parser_telemetry_tool::STRING_LEN_TYRE_COMPOUND));
}
frame.driver_name = string_cache.driver_name.clone();
frame.car_name = string_cache.car_name.clone();
frame.nationality = string_cache.nationality.clone();
frame.tyre_compound = string_cache.tyre_compound.clone();
Some(frame)
}
Err(_) => None,
}
};
// Update state in a single brief lock acquisition.
@ -344,7 +382,7 @@ async fn run_feed(
}
}
}
_ = shutdown.notified() => {
_ = shutdown_rx.changed() => {
break;
}
}
@ -391,7 +429,7 @@ async fn main() {
let mut feed_displays: Vec<(FeedDisplayInfo, Arc<std::sync::Mutex<FeedState>>)> = Vec::new();
let mut feed_tasks: Vec<tokio::task::JoinHandle<()>> = Vec::new();
let shutdown = Arc::new(tokio::sync::Notify::new());
let (shutdown_tx, shutdown_rx) = tokio::sync::watch::channel(false);
for feed in FEEDS {
let state = Arc::new(std::sync::Mutex::new(FeedState::new()));
@ -446,7 +484,7 @@ async fn main() {
if let Some(socket) = socket_opt {
let feed_clone = feed.clone();
let dir = session_dir.clone();
let sh = shutdown.clone();
let sh = shutdown_rx.clone();
feed_tasks.push(tokio::spawn(async move {
run_feed(feed_clone, socket, dir, state, sh).await;
}));
@ -507,7 +545,7 @@ async fn main() {
}
event_running.store(false, Ordering::Relaxed);
shutdown.notify_waiters();
let _ = shutdown_tx.send(true);
match tokio::time::timeout(
Duration::from_secs(3),

View file

@ -93,7 +93,7 @@ pub fn parse_rt_car_info(data: &[u8]) -> Result<TelemetryFrame, KsudpParseError>
// Driver inputs
throttle: read_f32(data, 51),
brake: read_f32(data, 55),
clutch: read_f32(data, 59),
clutch: 0.0, // offset 59 is fuel (0.0–1.0), not clutch; KSUDP has no clutch input
engine_rpm: read_f32(data, 63),
steer: read_f32(data, 67),

View file

@ -366,15 +366,6 @@ fn parse_v2_car_physics(data: &[u8]) -> Result<TelemetryFrame, PcarsParseError>
Ok(frame)
}
/// Read a u16 at the given offset (little-endian).
fn read_u16(data: &[u8], offset: usize) -> u16 {
let buf: [u8; 2] = data
.get(offset..offset + 2)
.and_then(|s| s.try_into().ok())
.unwrap_or([0u8; 2]);
u16::from_le_bytes(buf)
}
#[cfg(test)]
mod tests {
use super::*;

View file

@ -104,6 +104,7 @@ pub fn parse_packet(data: &[u8]) -> Result<TelemetryToolPacket, TelemetryToolPar
///
/// Avoids allocating a `Vec` for all drivers when only the local player's
/// telemetry is needed.
#[allow(dead_code)]
pub fn parse_first_driver(data: &[u8]) -> Result<TelemetryFrame, TelemetryToolParseError> {
if data.is_empty() {
return Err(TelemetryToolParseError::TooShort { actual: 0, needed: 1 });
@ -125,8 +126,56 @@ pub fn parse_first_driver(data: &[u8]) -> Result<TelemetryFrame, TelemetryToolPa
parse_driver(&payload[..DRIVER_STRUCT_SIZE])
}
/// Parse only the first driver, skipping rarely-changing string fields.
///
/// Returns a tuple of (frame, raw_driver_id). Strings can be populated
/// from a cache keyed by `raw_driver_id`.
pub fn parse_first_driver_no_strings(data: &[u8]) -> Result<(TelemetryFrame, u32), TelemetryToolParseError> {
if data.is_empty() {
return Err(TelemetryToolParseError::TooShort { actual: 0, needed: 1 });
}
let packet_type = data[0];
if packet_type != PACKET_TYPE_TELEMETRY {
return Err(TelemetryToolParseError::UnknownPacketType { got: packet_type });
}
let payload = &data[1..];
if payload.len() < DRIVER_STRUCT_SIZE {
return Err(TelemetryToolParseError::TooShort {
actual: data.len(),
needed: 1 + DRIVER_STRUCT_SIZE,
});
}
let driver_data = &payload[..DRIVER_STRUCT_SIZE];
let driver_id = read_u32(driver_data, 0);
let mut frame = parse_driver_core(driver_data)?;
frame.driver_id = Some(driver_id);
Ok((frame, driver_id))
}
pub const STRING_OFFSET_DRIVER_NAME: usize = 1 + 97;
pub const STRING_LEN_DRIVER_NAME: usize = 33;
pub const STRING_OFFSET_NATIONALITY: usize = 1 + 130;
pub const STRING_LEN_NATIONALITY: usize = 4;
pub const STRING_OFFSET_CAR_NAME: usize = 1 + 134;
pub const STRING_LEN_CAR_NAME: usize = 33;
pub const STRING_OFFSET_TYRE_COMPOUND: usize = 1 + 167;
pub const STRING_LEN_TYRE_COMPOUND: usize = 10;
/// Parse a single 357-byte driver structure.
fn parse_driver(data: &[u8]) -> Result<TelemetryFrame, TelemetryToolParseError> {
let mut frame = parse_driver_core(data)?;
frame.driver_id = Some(read_u32(data, 0));
frame.driver_name = Some(read_string(data, 97, 33));
frame.nationality = Some(read_string(data, 130, 4));
frame.car_name = Some(read_string(data, 134, 33));
frame.tyre_compound = Some(read_string(data, 167, 10));
Ok(frame)
}
fn parse_driver_core(data: &[u8]) -> Result<TelemetryFrame, TelemetryToolParseError> {
// Validation: check test_float at offset 93
let test_float_val = read_f32(data, 93);
if (test_float_val - TEST_FLOAT).abs() > 1.0 {
@ -195,11 +244,6 @@ fn parse_driver(data: &[u8]) -> Result<TelemetryFrame, TelemetryToolParseError>
tyre_slip: read_f32_4(data, 305),
// Telemetry Tool-specific fields
driver_id: Some(read_u32(data, 0)),
driver_name: Some(read_string(data, 97, 33)),
nationality: Some(read_string(data, 130, 4)),
car_name: Some(read_string(data, 134, 33)),
tyre_compound: Some(read_string(data, 167, 10)),
track_position: Some(read_i16(data, 67)),
leaderboard_position: Some(read_i16(data, 69)),
is_connected: Some(read_bool(data, 181)),

View file

@ -3,8 +3,6 @@
//! Both the KSUDP (port 9996) and Telemetry Tool (port 10101) parsers produce
//! `TelemetryFrame` instances from their raw binary data.
use std::fmt;
/// Identifies which feed source produced this frame.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum FeedSource {
@ -242,13 +240,3 @@ impl TelemetryFrame {
}
}
}
impl fmt::Display for FeedSource {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
FeedSource::Ksudp => write!(f, "KSUDP"),
FeedSource::TelemetryTool => write!(f, "Telemetry Tool"),
FeedSource::Pcars => write!(f, "PCARS"),
}
}
}