viva_gige/
message.rs

1//! GVCP message/event channel handling.
2//!
3//! A device delivers events to the controller as GVCP **commands** on the
4//! message channel — `EVENT_CMD` (0x00C0) for bare notifications and
5//! `EVENTDATA_CMD` (0x00C2) when the event carries device-specific data. Both
6//! are commands, so a datagram begins with the 0x42 key byte and a flags byte,
7//! not with a status code, and both may ask the controller to acknowledge.
8//!
9//! One `EVENT_CMD` can pack several events: the payload is an array of
10//! fixed-size entries, 16 bytes each with 16-bit block IDs or 24 bytes each
11//! when the device sets the extended-ID flag (GigE Vision 2.0). Layout per
12//! entry, from the GigE Vision field table and corroborated by Wireshark's
13//! `dissect_event_cmd`:
14//!
15//! ```text
16//! 16-bit block IDs (16 bytes)      64-bit block IDs (24 bytes)
17//!  0  reserved            u16       0  reserved            u16
18//!  2  event identifier    u16       2  event identifier    u16
19//!  4  stream channel      u16       4  stream channel      u16
20//!  6  block id            u16       6  reserved            u16
21//!  8  timestamp           u64       8  block id            u64
22//!                                  16  timestamp           u64
23//! ```
24
25use std::collections::VecDeque;
26use std::io;
27use std::io::ErrorKind;
28use std::net::{IpAddr, SocketAddr};
29
30use bytes::Bytes;
31#[cfg(test)]
32use bytes::{BufMut, BytesMut};
33use socket2::{Domain, Protocol, Socket, Type};
34use tokio::net::UdpSocket;
35use tokio::sync::Mutex;
36use tracing::{debug, info, trace, warn};
37
38use crate::gvcp::consts as gvcp;
39
40/// Constants related to GVCP message packets.
41mod consts {
42    /// Size of the GVCP message header in bytes.
43    pub const GVCP_HEADER: usize = 8;
44    /// Default receive buffer size requested for the UDP socket (bytes).
45    pub const DEFAULT_RCVBUF: usize = 1 << 20; // 1 MiB.
46    /// Maximum datagram size accepted on the event channel (bytes).
47    pub const MAX_EVENT_SIZE: usize = 2048;
48    /// GVCP command message key: the first byte of every GVCP *command*.
49    ///
50    /// Events arrive as commands from the device, not as acknowledgements —
51    /// which is why the first two bytes are the key and the flags, and not a
52    /// status code.
53    pub const GVCP_CMD_KEY: u8 = 0x42;
54    /// Flags-byte bit requesting an acknowledgement from the controller.
55    pub const FLAG_ACK_REQUIRED: u8 = 0x01;
56    /// Flags-byte bit marking 64-bit block IDs (GigE Vision 2.0).
57    pub const FLAG_EXTENDED_IDS: u8 = 0x10;
58}
59
60/// Header of a datagram received on the GVCP message channel.
61#[derive(Debug, Clone, Copy, PartialEq, Eq)]
62struct MessageHeader {
63    command: u16,
64    length: usize,
65    request_id: u16,
66    ack_required: bool,
67    extended_ids: bool,
68}
69
70impl MessageHeader {
71    fn parse(data: &[u8]) -> io::Result<Self> {
72        if data.len() < consts::GVCP_HEADER {
73            return Err(io::Error::new(ErrorKind::InvalidData, "packet too short"));
74        }
75        if data.len() > consts::MAX_EVENT_SIZE {
76            return Err(io::Error::new(ErrorKind::InvalidData, "packet too large"));
77        }
78        if data[0] != consts::GVCP_CMD_KEY {
79            return Err(io::Error::new(
80                ErrorKind::InvalidData,
81                "not a GVCP command packet",
82            ));
83        }
84        let flags = data[1];
85        let command = u16::from_be_bytes([data[2], data[3]]);
86        let length = u16::from_be_bytes([data[4], data[5]]) as usize;
87        let request_id = u16::from_be_bytes([data[6], data[7]]);
88
89        if !matches!(command, gvcp::EVENT_COMMAND | gvcp::EVENTDATA_COMMAND) {
90            return Err(io::Error::new(
91                ErrorKind::InvalidData,
92                "unexpected opcode for event packet",
93            ));
94        }
95        if length + consts::GVCP_HEADER != data.len() {
96            return Err(io::Error::new(ErrorKind::InvalidData, "length mismatch"));
97        }
98
99        Ok(Self {
100            command,
101            length,
102            request_id,
103            ack_required: flags & consts::FLAG_ACK_REQUIRED != 0,
104            extended_ids: flags & consts::FLAG_EXTENDED_IDS != 0,
105        })
106    }
107
108    /// Opcode of the acknowledgement this command expects.
109    fn ack_opcode(&self) -> u16 {
110        match self.command {
111            gvcp::EVENTDATA_COMMAND => gvcp::EVENTDATA_ACK,
112            _ => gvcp::EVENT_ACK,
113        }
114    }
115
116    /// Size of one event entry given the extended-ID flag.
117    fn entry_size(&self) -> usize {
118        if self.extended_ids {
119            gvcp::EVENT_ENTRY_EXTENDED
120        } else {
121            gvcp::EVENT_ENTRY
122        }
123    }
124}
125
126/// Parsed representation of a single GVCP event.
127#[derive(Debug, Clone, PartialEq, Eq)]
128pub struct EventPacket {
129    /// Source address of the datagram.
130    pub src: SocketAddr,
131    /// Event identifier reported by the device.
132    pub event_id: u16,
133    /// Device timestamp carried by the event (ticks).
134    pub timestamp_dev: u64,
135    /// Stream channel associated with the event.
136    pub stream_channel: u16,
137    /// GVSP block identifier associated with the event.
138    ///
139    /// 16-bit on GigE Vision 1.x devices, widened here so a 2.0 device using
140    /// extended block IDs fits without a second type.
141    pub block_id: u64,
142    /// Event data following the entry, empty for a bare `EVENT_CMD`.
143    pub payload: Bytes,
144}
145
146impl EventPacket {
147    /// Decode one event entry. `entry` must be at least `header.entry_size()`.
148    fn parse_entry(src: SocketAddr, header: &MessageHeader, entry: &[u8], payload: Bytes) -> Self {
149        let event_id = u16::from_be_bytes([entry[2], entry[3]]);
150        let stream_channel = u16::from_be_bytes([entry[4], entry[5]]);
151        let (block_id, ts_at) = if header.extended_ids {
152            // entry[6..8] is reserved in the extended layout.
153            let id = u64::from_be_bytes([
154                entry[8], entry[9], entry[10], entry[11], entry[12], entry[13], entry[14],
155                entry[15],
156            ]);
157            (id, 16)
158        } else {
159            (u64::from(u16::from_be_bytes([entry[6], entry[7]])), 8)
160        };
161        let timestamp_dev = u64::from_be_bytes([
162            entry[ts_at],
163            entry[ts_at + 1],
164            entry[ts_at + 2],
165            entry[ts_at + 3],
166            entry[ts_at + 4],
167            entry[ts_at + 5],
168            entry[ts_at + 6],
169            entry[ts_at + 7],
170        ]);
171        Self {
172            src,
173            event_id,
174            timestamp_dev,
175            stream_channel,
176            block_id,
177            payload,
178        }
179    }
180
181    /// Decode every event carried by one message-channel datagram.
182    fn parse_datagram(src: SocketAddr, data: &[u8]) -> io::Result<(MessageHeader, Vec<Self>)> {
183        let header = MessageHeader::parse(data)?;
184        let entry_size = header.entry_size();
185        let body = &data[consts::GVCP_HEADER..];
186
187        if body.len() < entry_size {
188            return Err(io::Error::new(
189                ErrorKind::InvalidData,
190                "event payload shorter than one entry",
191            ));
192        }
193
194        let events = if header.command == gvcp::EVENTDATA_COMMAND {
195            // One entry, then device-specific data to the end of the datagram.
196            // Packing several variable-length events into one EVENTDATA_CMD
197            // needs the GEV 2.1 per-event size field, which we do not read;
198            // no corpus device has been observed doing it.
199            let payload = Bytes::copy_from_slice(&body[entry_size..]);
200            vec![Self::parse_entry(src, &header, body, payload)]
201        } else {
202            if header.length % entry_size != 0 {
203                return Err(io::Error::new(
204                    ErrorKind::InvalidData,
205                    "event payload is not a whole number of entries",
206                ));
207            }
208            body.chunks_exact(entry_size)
209                .map(|entry| Self::parse_entry(src, &header, entry, Bytes::new()))
210                .collect()
211        };
212
213        Ok((header, events))
214    }
215}
216
217/// Build the acknowledgement for a message-channel command.
218///
219/// GVCP acknowledgements carry no payload here: status, opcode, a zero length
220/// and the request id the device used.
221fn encode_ack(ack_opcode: u16, request_id: u16) -> [u8; consts::GVCP_HEADER] {
222    let mut buf = [0u8; consts::GVCP_HEADER];
223    buf[0..2].copy_from_slice(&viva_gencp::StatusCode::Success.to_raw().to_be_bytes());
224    buf[2..4].copy_from_slice(&ack_opcode.to_be_bytes());
225    buf[4..6].copy_from_slice(&0u16.to_be_bytes());
226    buf[6..8].copy_from_slice(&request_id.to_be_bytes());
227    buf
228}
229
230/// Async GVCP message channel socket.
231pub struct EventSocket {
232    sock: UdpSocket,
233    buffer: Mutex<Vec<u8>>,
234    /// Events decoded from a multi-event datagram but not yet returned.
235    pending: Mutex<VecDeque<EventPacket>>,
236}
237
238impl EventSocket {
239    /// Bind a GVCP message socket on the provided local address.
240    pub async fn bind(local_ip: IpAddr, port: u16) -> io::Result<Self> {
241        let domain = match local_ip {
242            IpAddr::V4(_) => Domain::IPV4,
243            IpAddr::V6(_) => Domain::IPV6,
244        };
245        let socket = Socket::new(domain, Type::DGRAM, Some(Protocol::UDP))?;
246        socket.set_reuse_address(true)?;
247        socket.set_nonblocking(true)?;
248        if let Err(err) = socket.set_recv_buffer_size(consts::DEFAULT_RCVBUF) {
249            warn!(?err, "failed to grow GVCP message buffer");
250        }
251        let addr = SocketAddr::new(local_ip, port);
252        socket.bind(&addr.into())?;
253        let sock = UdpSocket::from_std(socket.into())?;
254        info!(local = %addr, "bound GVCP message socket");
255        Ok(Self {
256            sock,
257            buffer: Mutex::new(vec![0u8; consts::MAX_EVENT_SIZE]),
258            pending: Mutex::new(VecDeque::new()),
259        })
260    }
261
262    /// Receive and parse the next GVCP event.
263    ///
264    /// A datagram carrying several events is decoded once and drained across
265    /// successive calls. When the device set the acknowledge-required flag the
266    /// acknowledgement is sent before any of its events are returned — a device
267    /// that does not get one will retransmit.
268    ///
269    /// Concurrent callers are symmetric consumers of one queue, and none of
270    /// them is promised the event its own `recv_from` decoded.
271    pub async fn recv(&self) -> io::Result<EventPacket> {
272        loop {
273            if let Some(packet) = self.pending.lock().await.pop_front() {
274                return Ok(packet);
275            }
276
277            // Two locks, and only one order: the receive lock, then the queue.
278            // A caller blocked in `recv_from` must not hold the queue, or a
279            // second caller could not drain events that have already arrived.
280            let mut buffer = self.buffer.lock().await;
281            // Recheck: waiting for the receive lock is the window in which
282            // another caller queues a decoded datagram, and those events are
283            // older than anything the socket will hand us next.
284            if let Some(packet) = self.pending.lock().await.pop_front() {
285                return Ok(packet);
286            }
287
288            let (len, src) = self.sock.recv_from(&mut buffer[..]).await?;
289            trace!(bytes = len, %src, "received GVCP message");
290
291            match EventPacket::parse_datagram(src, &buffer[..len]) {
292                Ok((header, events)) => {
293                    if events.is_empty() {
294                        continue;
295                    }
296                    if header.ack_required {
297                        let ack = encode_ack(header.ack_opcode(), header.request_id);
298                        if let Err(err) = self.sock.send_to(&ack, src).await {
299                            warn!(%src, error = %err, "failed to acknowledge event");
300                        } else {
301                            trace!(%src, request_id = header.request_id, "acknowledged event");
302                        }
303                    }
304                    // Every event this datagram carried is published in one
305                    // step, while the receive lock is still held. Returning
306                    // one directly and queueing the remainder would be the
307                    // same code with a window in it: between releasing the
308                    // receive lock and queueing the rest, another caller can
309                    // take the lock, find the queue empty and block in
310                    // `recv_from` on a device that may never speak again,
311                    // while the events it wanted sat undelivered.
312                    debug!(events = events.len(), %src, "queueing GVCP events");
313                    self.pending.lock().await.extend(events);
314                }
315                Err(err) => {
316                    warn!(%src, error = %err, "discarding malformed event packet");
317                }
318            }
319        }
320    }
321
322    /// Return the local address bound to the socket.
323    pub fn local_addr(&self) -> io::Result<SocketAddr> {
324        self.sock.local_addr()
325    }
326
327    /// Access the underlying UDP socket (tests only).
328    #[cfg(test)]
329    pub fn socket(&self) -> &UdpSocket {
330        &self.sock
331    }
332}
333
334#[cfg(test)]
335mod tests {
336    use super::*;
337    use std::net::Ipv4Addr;
338    use std::sync::Arc;
339    use std::time::Duration;
340
341    fn src() -> SocketAddr {
342        SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 3956)
343    }
344
345    /// One `EVENT_CMD` with a single 16-byte entry, written out from the field
346    /// table rather than produced by our own encoder (ADR-0019).
347    ///
348    /// Every field in this fixture used to land somewhere else: the parser read
349    /// the event id out of the reserved word, the timestamp out of the stream
350    /// channel and block id, and rejected the packet outright because it
351    /// expected opcode 0x000D — which is not a GVCP opcode at all.
352    #[rustfmt::skip]
353    const EVENT_CMD_GOLDEN: [u8; 24] = [
354        0x42,                   // command key
355        0x01,                   // flags: acknowledge required
356        0x00, 0xC0,             // EVENT_CMD
357        0x00, 0x10,             // length: one 16-byte entry
358        0xCA, 0xFE,             // request id
359        0x00, 0x00,             // reserved
360        0x12, 0x34,             // event identifier
361        0x00, 0x07,             // stream channel index
362        0x00, 0x08,             // block id (16-bit)
363        0x00, 0x02, 0x00, 0x03, // timestamp, big-endian u64
364        0x00, 0x04, 0x00, 0x05,
365    ];
366
367    #[test]
368    fn event_cmd_matches_spec_offsets() {
369        let (header, events) =
370            EventPacket::parse_datagram(src(), &EVENT_CMD_GOLDEN).expect("parse");
371        assert_eq!(header.command, gvcp::EVENT_COMMAND);
372        assert!(header.ack_required);
373        assert!(!header.extended_ids);
374        assert_eq!(events.len(), 1);
375        let ev = &events[0];
376        assert_eq!(ev.event_id, 0x1234);
377        assert_eq!(ev.stream_channel, 7);
378        assert_eq!(ev.block_id, 8);
379        assert_eq!(ev.timestamp_dev, 0x0002_0003_0004_0005);
380        assert!(ev.payload.is_empty());
381    }
382
383    /// The opcode the parser used to demand. 0x000D is in the GenCP register
384    /// range, not the GVCP event range, so no device ever sent it.
385    #[test]
386    fn event_opcodes_are_in_the_gvcp_event_range() {
387        assert_eq!(gvcp::EVENT_COMMAND, 0x00C0);
388        assert_eq!(gvcp::EVENT_ACK, 0x00C1);
389        assert_eq!(gvcp::EVENTDATA_COMMAND, 0x00C2);
390        assert_eq!(gvcp::EVENTDATA_ACK, 0x00C3);
391
392        let mut wrong = EVENT_CMD_GOLDEN;
393        wrong[2..4].copy_from_slice(&0x000Du16.to_be_bytes());
394        assert!(EventPacket::parse_datagram(src(), &wrong).is_err());
395    }
396
397    #[test]
398    fn multiple_events_in_one_datagram_are_all_returned() {
399        let mut buf = BytesMut::new();
400        buf.put_u8(consts::GVCP_CMD_KEY);
401        buf.put_u8(0);
402        buf.put_u16(gvcp::EVENT_COMMAND);
403        buf.put_u16((gvcp::EVENT_ENTRY * 3) as u16);
404        buf.put_u16(0x0001);
405        for i in 0..3u16 {
406            buf.put_u16(0); // reserved
407            buf.put_u16(0x1000 + i); // event id
408            buf.put_u16(i); // stream channel
409            buf.put_u16(100 + i); // block id
410            buf.put_u64(u64::from(i) + 1);
411        }
412        let (_, events) = EventPacket::parse_datagram(src(), &buf).expect("parse");
413        assert_eq!(events.len(), 3);
414        assert_eq!(
415            events.iter().map(|e| e.event_id).collect::<Vec<_>>(),
416            vec![0x1000, 0x1001, 0x1002]
417        );
418        assert_eq!(events[2].block_id, 102);
419        assert_eq!(events[2].timestamp_dev, 3);
420    }
421
422    #[test]
423    fn extended_block_ids_shift_the_timestamp() {
424        let mut buf = BytesMut::new();
425        buf.put_u8(consts::GVCP_CMD_KEY);
426        buf.put_u8(consts::FLAG_EXTENDED_IDS);
427        buf.put_u16(gvcp::EVENT_COMMAND);
428        buf.put_u16(gvcp::EVENT_ENTRY_EXTENDED as u16);
429        buf.put_u16(0x0002);
430        buf.put_u16(0); // reserved
431        buf.put_u16(0x4321); // event id
432        buf.put_u16(3); // stream channel
433        buf.put_u16(0); // reserved
434        buf.put_u64(0x0102_0304_0506_0708); // 64-bit block id
435        buf.put_u64(0x1122_3344_5566_7788); // timestamp
436
437        let (header, events) = EventPacket::parse_datagram(src(), &buf).expect("parse");
438        assert!(header.extended_ids);
439        assert_eq!(events[0].event_id, 0x4321);
440        assert_eq!(events[0].block_id, 0x0102_0304_0506_0708);
441        assert_eq!(events[0].timestamp_dev, 0x1122_3344_5566_7788);
442    }
443
444    #[test]
445    fn eventdata_carries_a_payload() {
446        let data = [0xAAu8, 0xBB, 0xCC, 0xDD];
447        let mut buf = BytesMut::new();
448        buf.put_u8(consts::GVCP_CMD_KEY);
449        buf.put_u8(0);
450        buf.put_u16(gvcp::EVENTDATA_COMMAND);
451        buf.put_u16((gvcp::EVENT_ENTRY + data.len()) as u16);
452        buf.put_u16(0x0003);
453        buf.put_u16(0);
454        buf.put_u16(0x0009);
455        buf.put_u16(1);
456        buf.put_u16(42);
457        buf.put_u64(0xDEAD_BEEF);
458        buf.extend_from_slice(&data);
459
460        let (header, events) = EventPacket::parse_datagram(src(), &buf).expect("parse");
461        assert_eq!(header.ack_opcode(), gvcp::EVENTDATA_ACK);
462        assert_eq!(events.len(), 1);
463        assert_eq!(events[0].event_id, 0x0009);
464        assert_eq!(events[0].block_id, 42);
465        assert_eq!(&events[0].payload[..], &data);
466    }
467
468    #[test]
469    fn ack_matches_spec_bytes() {
470        let ack = encode_ack(gvcp::EVENT_ACK, 0xCAFE);
471        assert_eq!(ack, [0x00, 0x00, 0x00, 0xC1, 0x00, 0x00, 0xCA, 0xFE]);
472    }
473
474    /// Two concurrent receivers must share a multi-event datagram.
475    ///
476    /// Both tasks start while the queue is empty, so both get past the fast
477    /// path and one blocks on the receive lock. When the datagram lands, the
478    /// winner queues both events, and the loser wakes holding a lock it must
479    /// not carry into `recv_from` — the device has already said everything it
480    /// is going to say. Without the recheck after acquiring the receive lock
481    /// this hangs with an event sitting in the queue.
482    ///
483    /// It does *not* cover the narrower ordering the recheck depends on:
484    /// publishing the decoded events before releasing the receive lock. That
485    /// window is a few instructions wide, so no timing test holds it open
486    /// reliably — which is why the events are published in one step inside the
487    /// lock rather than split into a returned first and a queued remainder.
488    #[tokio::test(flavor = "multi_thread")]
489    async fn a_queued_event_is_not_stranded_behind_the_receive_lock() {
490        let sock = Arc::new(
491            EventSocket::bind(IpAddr::V4(Ipv4Addr::LOCALHOST), 0)
492                .await
493                .expect("bind"),
494        );
495        let dest = sock.local_addr().expect("local addr");
496
497        let first = tokio::spawn({
498            let sock = Arc::clone(&sock);
499            async move { sock.recv().await.expect("first event") }
500        });
501        let second = tokio::spawn({
502            let sock = Arc::clone(&sock);
503            async move { sock.recv().await.expect("second event") }
504        });
505
506        // Let both tasks reach the receive lock before any data exists.
507        tokio::time::sleep(Duration::from_millis(50)).await;
508
509        let mut buf = BytesMut::new();
510        buf.put_u8(consts::GVCP_CMD_KEY);
511        // Ask for an acknowledgement, as a device wanting delivery confirmed
512        // does: the ack is sent while the receive lock is held, so this also
513        // covers that sending it does not stall the second consumer.
514        buf.put_u8(consts::FLAG_ACK_REQUIRED);
515        buf.put_u16(gvcp::EVENT_COMMAND);
516        buf.put_u16((gvcp::EVENT_ENTRY * 2) as u16);
517        buf.put_u16(0x0001);
518        for i in 0..2u16 {
519            buf.put_u16(0); // reserved
520            buf.put_u16(0x2000 + i); // event id
521            buf.put_u16(0); // stream channel
522            buf.put_u16(i); // block id
523            buf.put_u64(u64::from(i));
524        }
525        let sender = UdpSocket::bind("127.0.0.1:0").await.expect("sender");
526        sender.send_to(&buf, dest).await.expect("send");
527
528        let both = tokio::time::timeout(Duration::from_secs(5), async {
529            (first.await.expect("join"), second.await.expect("join"))
530        })
531        .await
532        .expect("both receivers finished");
533
534        let mut ids = [both.0.event_id, both.1.event_id];
535        ids.sort_unstable();
536        assert_eq!(ids, [0x2000, 0x2001]);
537    }
538
539    #[test]
540    fn reject_short_packet() {
541        let err = EventPacket::parse_datagram(src(), &[0x42, 0x00, 0x00]).unwrap_err();
542        assert_eq!(err.kind(), ErrorKind::InvalidData);
543    }
544
545    #[test]
546    fn reject_ack_shaped_packet() {
547        // An acknowledgement, not a command: no 0x42 key byte.
548        let mut buf = EVENT_CMD_GOLDEN;
549        buf[0] = 0x00;
550        assert!(EventPacket::parse_datagram(src(), &buf).is_err());
551    }
552
553    #[test]
554    fn reject_partial_entry() {
555        let mut buf = EVENT_CMD_GOLDEN.to_vec();
556        buf.truncate(consts::GVCP_HEADER + 12);
557        buf[4..6].copy_from_slice(&12u16.to_be_bytes());
558        assert!(EventPacket::parse_datagram(src(), &buf).is_err());
559    }
560}
viva_gige/message.rs

viva_gige/
message.rs
