use std::fmt;
use tracing::error;
pub const DEFAULT_CAPACITY: usize = 256 * 1024;
pub const SHARED_HEADER_SIZE: usize = 8;
const CMD_SIZE: usize = std::mem::size_of::<PluginPaintCmd>();
#[repr(C)]
#[derive(Copy, Clone, Debug, Default, PartialEq, Eq)]
pub struct PluginPaintCmd {
pub cmd_type: u16,
pub flags: u16,
pub data_len: u32,
pub payload_offset: u32,
}
impl PluginPaintCmd {
#[inline]
pub const fn header_size() -> usize {
CMD_SIZE
}
fn write_to(&self, buf: &mut [u8]) -> Option<()> {
if buf.len() < CMD_SIZE {
return None;
}
let (cmd_type, rest) = buf.split_at_mut(2);
cmd_type.copy_from_slice(&self.cmd_type.to_le_bytes());
let (flags, rest) = rest.split_at_mut(2);
flags.copy_from_slice(&self.flags.to_le_bytes());
let (data_len, rest) = rest.split_at_mut(4);
data_len.copy_from_slice(&self.data_len.to_le_bytes());
let (payload_offset, _) = rest.split_at_mut(4);
payload_offset.copy_from_slice(&self.payload_offset.to_le_bytes());
Some(())
}
fn read_from(buf: &[u8]) -> Option<Self> {
if buf.len() < CMD_SIZE {
return None;
}
let cmd_type = u16::from_le_bytes([buf[0], buf[1]]);
let flags = u16::from_le_bytes([buf[2], buf[3]]);
let data_len = u32::from_le_bytes([buf[4], buf[5], buf[6], buf[7]]);
let payload_offset = u32::from_le_bytes([buf[8], buf[9], buf[10], buf[11]]);
Some(Self {
cmd_type,
flags,
data_len,
payload_offset,
})
}
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum RingBufferError {
BufferFull,
PayloadLengthMismatch,
InvalidPayloadOffset,
}
impl fmt::Display for RingBufferError {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
RingBufferError::BufferFull => write!(f, "ring buffer is full"),
RingBufferError::PayloadLengthMismatch => {
write!(f, "command data_len does not match payload length")
}
RingBufferError::InvalidPayloadOffset => {
write!(f, "command payload offset or length is out of bounds")
}
}
}
}
impl std::error::Error for RingBufferError {}
pub struct PluginRingBuffer<'a> {
data: &'a mut [u8],
header: Option<&'a mut [u8]>,
head: u32,
tail: u32,
}
impl<'a> PluginRingBuffer<'a> {
pub fn new(data: &'a mut [u8]) -> Self {
Self {
data,
header: None,
head: 0,
tail: 0,
}
}
pub fn new_shared(data: &'a mut [u8]) -> Self {
if data.len() < SHARED_HEADER_SIZE {
return Self::new(data);
}
let (header, payload) = data.split_at_mut(SHARED_HEADER_SIZE);
let head = u32::from_le_bytes([header[0], header[1], header[2], header[3]]);
let tail = u32::from_le_bytes([header[4], header[5], header[6], header[7]]);
let capacity = payload.len() as u32;
let (head, tail) = if head <= capacity && tail <= capacity {
(head, tail)
} else {
(0, 0)
};
let mut rb = Self {
data: payload,
header: Some(header),
head,
tail,
};
rb.sync_header();
rb
}
fn sync_header(&mut self) {
if let Some(header) = self.header.as_deref_mut() {
header[0..4].copy_from_slice(&self.head.to_le_bytes());
header[4..8].copy_from_slice(&self.tail.to_le_bytes());
}
}
pub fn capacity(&self) -> usize {
self.data.len()
}
fn capacity_u32(&self) -> u32 {
self.data.len() as u32
}
pub fn len(&self) -> usize {
let cap = self.capacity_u32();
if self.head == self.tail {
0
} else if self.tail > self.head {
(self.tail - self.head) as usize
} else {
(cap - self.head + self.tail) as usize
}
}
pub fn is_empty(&self) -> bool {
self.head == self.tail
}
#[inline]
const fn record_size(payload_len: usize) -> usize {
CMD_SIZE.saturating_add(payload_len)
}
pub fn produce(&mut self, cmd: &PluginPaintCmd, payload: &[u8]) -> Result<(), RingBufferError> {
if cmd.data_len as usize != payload.len() {
return Err(RingBufferError::PayloadLengthMismatch);
}
let payload_len = payload.len();
let total = Self::record_size(payload_len);
if total == 0 || self.capacity() == 0 {
return Err(RingBufferError::BufferFull);
}
if self.len().saturating_add(total) >= self.capacity() {
return Err(RingBufferError::BufferFull);
}
let cap_u32 = self.capacity_u32();
let tail = self.tail as usize;
let (write_pos, wrapped) = if self.tail < self.head {
if tail.saturating_add(total) > self.head as usize {
return Err(RingBufferError::BufferFull);
}
(tail, false)
} else if tail.saturating_add(total) <= self.capacity() {
(tail, false)
} else if total < self.head as usize {
(0, true)
} else {
return Err(RingBufferError::BufferFull);
};
if wrapped && self.tail == self.head {
self.head = 0;
}
let payload_offset = write_pos + CMD_SIZE;
let mut stored_cmd = *cmd;
stored_cmd.payload_offset = payload_offset as u32;
stored_cmd
.write_to(&mut self.data[write_pos..])
.ok_or(RingBufferError::BufferFull)?;
self.data[payload_offset..payload_offset.saturating_add(payload_len)]
.copy_from_slice(payload);
self.tail = (write_pos + total) as u32;
if self.tail >= cap_u32 {
self.tail = 0;
}
self.sync_header();
Ok(())
}
pub fn consume(&mut self) -> Option<(PluginPaintCmd, &[u8])> {
if self.is_empty() {
return None;
}
let cap_u32 = self.capacity_u32();
if self.head >= cap_u32 {
self.head = 0;
if self.is_empty() {
self.sync_header();
return None;
}
}
let pos = self.head as usize;
if pos.saturating_add(CMD_SIZE) > self.data.len() {
self.head = 0;
return self.consume();
}
let cmd = match PluginPaintCmd::read_from(&self.data[pos..]) {
Some(cmd) => cmd,
None => {
error!(
head = self.head,
tail = self.tail,
pos,
"ring buffer: malformed command header; resetting cursors"
);
self.head = 0;
self.tail = 0;
self.sync_header();
return None;
}
};
let payload_len = cmd.data_len as usize;
let expected_payload_start = pos.saturating_add(CMD_SIZE);
let expected_payload_end = expected_payload_start.saturating_add(payload_len);
if cmd.payload_offset as usize != expected_payload_start
|| expected_payload_end > self.data.len()
|| expected_payload_end < expected_payload_start
{
error!(
head = self.head,
tail = self.tail,
pos,
payload_offset = cmd.payload_offset,
data_len = cmd.data_len,
expected_payload_start,
expected_payload_end,
buf_len = self.data.len(),
"ring buffer: malformed record (payload not contiguous); resetting cursors"
);
self.head = 0;
self.tail = 0;
self.sync_header();
return None;
}
self.head = (expected_payload_end) as u32;
if self.head >= cap_u32 {
self.head = 0;
}
if let Some(header) = self.header.as_deref_mut() {
header[0..4].copy_from_slice(&self.head.to_le_bytes());
header[4..8].copy_from_slice(&self.tail.to_le_bytes());
}
let payload = &self.data[expected_payload_start..expected_payload_end];
Some((cmd, payload))
}
pub fn drain<F>(&mut self, mut f: F)
where
F: FnMut(&PluginPaintCmd, &[u8]),
{
while let Some((cmd, payload)) = self.consume() {
f(&cmd, payload);
}
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn empty_buffer_returns_none() {
let mut data = vec![0u8; DEFAULT_CAPACITY];
let mut rb = PluginRingBuffer::new(&mut data);
assert!(rb.is_empty());
assert_eq!(rb.len(), 0);
assert!(rb.consume().is_none());
}
#[test]
fn produce_and_consume_single_command() {
let mut data = vec![0u8; DEFAULT_CAPACITY];
let mut rb = PluginRingBuffer::new(&mut data);
let cmd = PluginPaintCmd {
cmd_type: 1,
flags: 0,
data_len: 4,
payload_offset: 0,
};
rb.produce(&cmd, &[1, 2, 3, 4]).unwrap();
assert_eq!(rb.len(), CMD_SIZE + 4);
let (read_cmd, payload) = rb.consume().unwrap();
assert_eq!(read_cmd.cmd_type, 1);
assert_eq!(read_cmd.data_len, 4);
assert_eq!(payload, &[1, 2, 3, 4]);
assert!(rb.is_empty());
}
#[test]
fn payload_length_mismatch_is_rejected() {
let mut data = vec![0u8; DEFAULT_CAPACITY];
let mut rb = PluginRingBuffer::new(&mut data);
let cmd = PluginPaintCmd {
cmd_type: 1,
flags: 0,
data_len: 10,
payload_offset: 0,
};
assert_eq!(
rb.produce(&cmd, &[1, 2, 3, 4]),
Err(RingBufferError::PayloadLengthMismatch)
);
}
#[test]
fn wrap_around_reuses_start_of_buffer() {
let mut data = vec![0u8; 64];
let mut rb = PluginRingBuffer::new(&mut data);
let cmd = PluginPaintCmd {
cmd_type: 2,
flags: 0,
data_len: 40,
payload_offset: 0,
};
rb.produce(&cmd, &[7; 40]).unwrap();
rb.consume().unwrap();
let cmd2 = PluginPaintCmd {
cmd_type: 3,
flags: 0,
data_len: 16,
payload_offset: 0,
};
rb.produce(&cmd2, &[8; 16]).unwrap();
let (read_cmd, payload) = rb.consume().unwrap();
assert_eq!(read_cmd.cmd_type, 3);
assert_eq!(payload, &[8; 16]);
assert!(rb.is_empty());
}
#[test]
fn buffer_full_is_reported() {
let mut data = vec![0u8; 64];
let mut rb = PluginRingBuffer::new(&mut data);
let cmd = PluginPaintCmd {
cmd_type: 1,
flags: 0,
data_len: 40,
payload_offset: 0,
};
rb.produce(&cmd, &[1; 40]).unwrap();
assert_eq!(rb.produce(&cmd, &[1; 40]), Err(RingBufferError::BufferFull));
}
#[test]
fn drain_visits_all_commands() {
let mut data = vec![0u8; DEFAULT_CAPACITY];
let mut rb = PluginRingBuffer::new(&mut data);
let cmd = PluginPaintCmd {
cmd_type: 1,
flags: 0,
data_len: 2,
payload_offset: 0,
};
for i in 0u8..5 {
rb.produce(&cmd, &[i, i + 1]).unwrap();
}
let mut count = 0;
rb.drain(|_cmd, payload| {
assert_eq!(payload.len(), 2);
count += 1;
});
assert_eq!(count, 5);
assert!(rb.is_empty());
}
#[test]
fn cmd_serialization_roundtrips() {
let original = PluginPaintCmd {
cmd_type: 0xABCD,
flags: 0x1234,
data_len: 0xDEAD_BEEF,
payload_offset: 0xCAFE_BABE,
};
let mut buf = [0u8; CMD_SIZE];
original.write_to(&mut buf).unwrap();
let parsed = PluginPaintCmd::read_from(&buf).unwrap();
assert_eq!(original, parsed);
}
#[test]
fn zero_payload_command_roundtrips() {
let mut data = vec![0u8; 64];
let mut rb = PluginRingBuffer::new(&mut data);
let cmd = PluginPaintCmd {
cmd_type: 0,
flags: 0,
data_len: 0,
payload_offset: 0,
};
rb.produce(&cmd, &[]).unwrap();
let (read_cmd, payload) = rb.consume().unwrap();
assert_eq!(read_cmd.cmd_type, cmd.cmd_type);
assert_eq!(read_cmd.flags, cmd.flags);
assert_eq!(read_cmd.data_len, cmd.data_len);
assert!(payload.is_empty());
}
#[test]
fn malformed_record_resets_cursors_and_does_not_wedge() {
let mut data = vec![0u8; 64];
let bad = PluginPaintCmd {
cmd_type: 1,
flags: 0,
data_len: 4,
payload_offset: 0, };
bad.write_to(&mut data[0..]).unwrap();
data[CMD_SIZE..CMD_SIZE + 4].fill(9);
let mut rb = PluginRingBuffer::new(&mut data);
rb.head = 0;
rb.tail = (CMD_SIZE + 4) as u32;
assert!(rb.consume().is_none());
assert!(rb.is_empty());
assert!(rb.consume().is_none());
let cmd = PluginPaintCmd {
cmd_type: 2,
flags: 0,
data_len: 2,
payload_offset: 0,
};
rb.produce(&cmd, &[3, 4]).unwrap();
let (read_cmd, payload) = rb.consume().unwrap();
assert_eq!(read_cmd.cmd_type, 2);
assert_eq!(payload, &[3, 4]);
}
#[test]
fn malformed_record_resets_shared_header() {
use crate::ring_buffer::SHARED_HEADER_SIZE;
let mut region = vec![0u8; SHARED_HEADER_SIZE + 64];
{
let mut rb = PluginRingBuffer::new_shared(&mut region);
let cmd = PluginPaintCmd {
cmd_type: 1,
flags: 0,
data_len: 4,
payload_offset: 0,
};
rb.produce(&cmd, &[1, 2, 3, 4]).unwrap();
}
let off = SHARED_HEADER_SIZE + 8;
region[off..off + 4].copy_from_slice(&99u32.to_le_bytes());
{
let mut rb = PluginRingBuffer::new_shared(&mut region);
assert!(rb.consume().is_none());
}
let head = u32::from_le_bytes(region[0..4].try_into().unwrap());
let tail = u32::from_le_bytes(region[4..8].try_into().unwrap());
assert_eq!(head, 0);
assert_eq!(tail, 0);
}
}