pipeline GW: cleanup

proto FW: fix compilation error

src/gateware/cxp.py

@@ -25,14 +25,20 @@ class CXP_Interface(Module, AutoCSR):
     def get_tx_port(self):
         return self.upconn.command.mem.get_port(write_capable=True)
 
+    def get_tx_mem_size(self):
+        return self.upconn.command.mem.depth*self.upconn.command.mem.width
+
     def get_rx_port(self):
         return self.downconn.packet_decoder.mem.get_port(write_capable=False)
 
+    def get_rx_mem_size(self):
+        return self.downconn.packet_decoder.mem.depth*self.downconn.packet_decoder.mem.width
+
     def get_loopback_tx_port(self):
         return self.downconn.command.mem.get_port(write_capable=True)
     
-    def get_mem_size(self):
+    def get_loopback_tx_mem_size(self):
-        return buffer_depth*word_dw
+        return self.downconn.command.mem.depth*self.downconn.command.mem.width
 
 class DownConn_Interface(Module, AutoCSR):
     def __init__(self, phy, debug_sma, pmod_pads):
@@ -203,24 +209,20 @@ class DownConn_Interface(Module, AutoCSR):
         # Priority level 2 packet - data, test packet
         self.submodules.packet_decoder = packet_decoder = cdr(CXP_Data_Packet_Decode())
 
-        self.new_rx_packet = CSR()
         self.decoder_error = CSR()
         self.test_error = CSR()
+        self.buffer_error = CSR()
 
-        self.submodules.new_packet_ps = new_packet_ps = PulseSynchronizer("cxp_gtx_rx", "sys")
+        decode_err_ps = PulseSynchronizer("cxp_gtx_rx", "sys")
-        self.submodules.decode_err_ps = decode_err_ps = PulseSynchronizer("cxp_gtx_rx", "sys")
+        test_err_ps = PulseSynchronizer("cxp_gtx_rx", "sys")
-        self.submodules.test_err_ps = test_err_ps = PulseSynchronizer("cxp_gtx_rx", "sys")
+        buffer_err_ps = PulseSynchronizer("cxp_gtx_rx", "sys")
+        self.submodules += decode_err_ps, test_err_ps, buffer_err_ps
         self.comb += [
-            new_packet_ps.i.eq(packet_decoder.new_packet),
             decode_err_ps.i.eq(packet_decoder.decode_err),
             test_err_ps.i.eq(packet_decoder.test_err),
+            buffer_err_ps.i.eq(packet_decoder.buffer_err),
         ]
         self.sync += [
-            If(new_packet_ps.o,
-                self.new_rx_packet.w.eq(1),
-            ).Elif(self.new_rx_packet.re,
-                self.new_rx_packet.w.eq(0),
-            ),
             If(decode_err_ps.o,
                 self.decoder_error.w.eq(1),
             ).Elif(self.decoder_error.re,
@@ -231,17 +233,30 @@ class DownConn_Interface(Module, AutoCSR):
             ).Elif(self.test_error.re,
                 self.test_error.w.eq(0),
             ),
+            If(buffer_err_ps.o,
+                self.buffer_error.w.eq(1),
+            ).Elif(self.test_error.re,
+                self.buffer_error.w.eq(0),
+            ),
         ]
 
 
         # Cicular buffer interface
         self.packet_type = CSRStatus(8)
-        self.write_pointer = CSRStatus(bits_for(buffer_depth)) # for firmware to sync with buffer
+        self.pending_packet = CSR()
+        self.read_ptr = CSRStatus(log2_int(buffer_count))
 
         self.specials += [
             MultiReg(packet_decoder.packet_type, self.packet_type.status),
-            MultiReg(packet_decoder.write_ptr, self.write_pointer.status),
+            MultiReg(self.read_ptr.status, packet_decoder.read_ptr_rx, odomain="cxp_gtx_rx"),
         ]
+        self.sync += [
+            self.pending_packet.w.eq(self.read_ptr.status != packet_decoder.write_ptr_sys),
+            If(self.pending_packet.re & self.pending_packet.w, 
+                self.read_ptr.status.eq(self.read_ptr.status + 1),
+            )
+        ]
+        
 
         # DEBUG: remove this cdc fifo
         cdc_fifo = stream.AsyncFIFO(word_layout, 512)

src/gateware/cxp_pipeline.py

@@ -1,13 +1,17 @@
 from migen import *
+from migen.genlib.cdc import MultiReg 
 from misoc.interconnect.csr import *
 from misoc.interconnect import stream
+from misoc.cores.liteeth_mini.mac.crc import LiteEthMACCRCEngine, LiteEthMACCRCChecker
 
 char_width = 8
 char_layout = [("data", char_width), ("k", char_width//8)]
 
 word_dw = 32
 word_layout = [("data", word_dw), ("k", word_dw//8)]
-buffer_depth = 128
+
+buffer_count = 4
+buffer_depth = 512
 
 def K(x, y):
     return ((y << 5) | x)
@@ -178,7 +182,7 @@ class Trigger_ACK_Inserter(Module):
 @FullMemoryWE()
 class TX_Command_Packet(Module, AutoCSR):
     def __init__(self):
-        self.tx_word_len = CSRStorage(bits_for(buffer_depth))
+        self.tx_word_len = CSRStorage(log2_int(buffer_depth))
         self.tx = CSR()
 
         # # #
@@ -189,7 +193,7 @@ class TX_Command_Packet(Module, AutoCSR):
 
 
         tx_done = Signal()
-        addr_next = Signal(bits_for(buffer_depth))
+        addr_next = Signal(log2_int(buffer_depth))
         addr = Signal.like(addr_next)
         addr_rst = Signal()
         addr_inc = Signal()
@@ -246,7 +250,7 @@ class TX_Test_Packet(Module, AutoCSR):
         self.source = stream.Endpoint(word_layout)
         self.submodules.fsm = fsm = FSM(reset_state="IDLE")
         
-        cnt = Signal(0xFFF)
+        cnt = Signal(max=0xFFF)
         fsm.act("IDLE",
             NextValue(cnt, cnt.reset),
             If(self.tx.re,
@@ -297,20 +301,35 @@ class RX_Debug_Buffer(Module,AutoCSR):
             self.dout_valid.status.eq(buf_out.source.stb),
         ]
 
+class Duplicate_Majority_Voter(Module):
+    def __init__(self, data, k):
+        assert data.nbits == 32
+        assert k.nbits == 4
+        
+        # Section 9.2.2.1 (CXP-001-2021)
+        # decoder should immune to single bit errors when handling duplicated characters
+        self.char = Signal(char_width)
+        self.k = Signal()
+
+        a, a_k = data[:8], k[0]
+        b, b_k = data[8:16], k[1]
+        c, c_k = data[16:24], k[2]
+        d, d_k = data[24:], k[3]
+        self.comb += [
+            self.char.eq(a&b&c | a&b&d | a&c&d | b&c&d),
+            self.k.eq(a_k&b_k&c_k | a_k&b_k&d_k | a_k&c_k&d_k | b_k&c_k&d_k),
+        ]
+
 @FullMemoryWE()
 class CXP_Data_Packet_Decode(Module):
     def __init__(self):
         self.packet_type = Signal(8)
-        self.write_ptr = Signal(bits_for(buffer_depth))
 
-        self.new_packet = Signal()
         self.decode_err = Signal()
         self.test_err = Signal()
+        self.buffer_err = Signal()
         # # #
 
-        # DEBUG: remove debug
-        # TODO: decode all packet type here
-        # TODO: data&event -> memory
         # TODO: heartbeat
         type = {
             "data_stream": 0x01,
@@ -319,41 +338,47 @@ class CXP_Data_Packet_Decode(Module):
             "control_ack_with_tag": 0x06,
             "event": 0x07,
             "heartbeat": 0x09,
-
-            "debug" : 0x02,
         }
 
         self.sink = stream.Endpoint(word_layout)
         self.source = stream.Endpoint(word_layout)
 
         self.submodules.fsm = fsm = FSM(reset_state="IDLE")
+        self.submodules.voter = voter = Duplicate_Majority_Voter(self.sink.data, self.sink.k)
 
         fsm.act("IDLE",
             self.sink.ack.eq(1),
-            # TODO: add error correction?
+            If((self.sink.stb & (voter.char == KCode["pak_start"]) & (voter.k == 1)),
-            If((self.sink.stb & (self.sink.data == Replicate(KCode["pak_start"], 4)) & (self.sink.k == 0b1111)),
                 NextState("DECODE"),
             )
         )
 
         cnt = Signal(max=0x100)
+        addr_nbits = log2_int(buffer_depth)
+        addr = Signal(addr_nbits)
         fsm.act("DECODE",
             self.sink.ack.eq(1),
             If(self.sink.stb,
-                self.new_packet.eq(1),
+                NextValue(self.packet_type, voter.char),
-                NextValue(self.packet_type, self.sink.data[:8]),
 
-                Case(self.sink.data[:8],{
+                Case(voter.char, {
                     type["data_stream"]: NextState("STREAMING"),
-                    type["control_ack_no_tag"]: NextState("LOAD_BUFFER"),
                     type["test_packet"]: [
                         NextValue(cnt, cnt.reset),
                         NextState("VERIFY_TEST_PATTERN"),
                     ],
-                    type["control_ack_with_tag"]: NextState("LOAD_BUFFER"),
+                    type["control_ack_no_tag"]:[
-                    type["event"]: NextState("LOAD_BUFFER"),
+                        NextValue(addr, addr.reset),
-
+                        NextState("LOAD_BUFFER"),
-                    type["debug"]: NextState("LOAD_BUFFER"),
+                    ],
+                    type["control_ack_with_tag"]:[
+                        NextValue(addr, addr.reset),
+                        NextState("LOAD_BUFFER"),
+                    ],
+                    type["event"]: [
+                        NextValue(addr, addr.reset),
+                        NextState("LOAD_BUFFER"),
+                    ],
                     "default": [
                          self.decode_err.eq(1),
                          # wait till next valid packet
@@ -362,6 +387,16 @@ class CXP_Data_Packet_Decode(Module):
                 }),
             )
         )
+        # For stream data packet
+        fsm.act("STREAMING",
+            If((self.sink.stb & (voter.char == KCode["pak_end"]) & (voter.k == 1)),
+                # discard K29,7
+                self.sink.ack.eq(1),
+                NextState("IDLE")
+            ).Else(
+                self.sink.connect(self.source),
+            )      
+        )
 
         # Section 9.9.1 (CXP-001-2021)
         # the received test data packet (0x00, 0x01 ... 0xFF) 
@@ -369,7 +404,7 @@ class CXP_Data_Packet_Decode(Module):
         fsm.act("VERIFY_TEST_PATTERN",
             self.sink.ack.eq(1),
             If(self.sink.stb,
-                If(((self.sink.data == Replicate(KCode["pak_end"], 4)) & (self.sink.k == 0b1111)),
+                If(((voter.char == KCode["pak_end"]) & (voter.k == 1)),
                     NextState("IDLE"),
                 ).Else(
                     If(((self.sink.data != Cat(cnt, cnt+1, cnt+2, cnt+3))),  
@@ -385,38 +420,52 @@ class CXP_Data_Packet_Decode(Module):
                 
         )
 
-        # For stream data packet
-        fsm.act("STREAMING",
-            If((self.sink.stb & (self.sink.data == Replicate(KCode["pak_end"], 4)) & (self.sink.k == 0b1111)),
-                # discard K29,7
-                self.sink.ack.eq(1),
-                NextState("IDLE")
-            ).Else(
-                self.sink.connect(self.source),
-            )      
-        )
-
         # A circular buffer for firmware to read packet from
-        self.specials.mem = mem = Memory(word_dw, buffer_depth)
+        self.specials.mem = mem = Memory(word_dw, buffer_count*buffer_depth)
         self.specials.mem_port = mem_port = mem.get_port(write_capable=True)
 
-        self.comb += mem_port.adr.eq(self.write_ptr),
+        write_ptr = Signal(log2_int(buffer_count))
+        self.write_ptr_sys = Signal.like(write_ptr)
+        self.specials += MultiReg(write_ptr, self.write_ptr_sys),
+
+        self.comb += [
+            mem_port.adr[:addr_nbits].eq(addr),
+            mem_port.adr[addr_nbits:].eq(write_ptr),
+        ]
 
         # For control ack, event packet
         fsm.act("LOAD_BUFFER",
             mem_port.we.eq(0),
             self.sink.ack.eq(1),
             If(self.sink.stb,
-                If(((self.sink.data == Replicate(KCode["pak_end"], 4)) & (self.sink.k == 0b1111)),
+                If(((voter.char == KCode["pak_end"]) & (voter.k == 1)),
-                    NextState("IDLE"),
+                    NextState("MOVE_BUFFER_PTR"),
                 ).Else(
                     mem_port.we.eq(1),
                     mem_port.dat_w.eq(self.sink.data),
-                    NextValue(self.write_ptr, self.write_ptr + 1),
+                    NextValue(addr, addr + 1),
+
+                    If(addr == buffer_depth - 1,
+                        # discard the packet
+                        self.buffer_err.eq(1),
+                        NextState("IDLE"),
+                    )
                 )
             )
         )
 
+        self.read_ptr_rx = Signal.like(write_ptr)
+        fsm.act("MOVE_BUFFER_PTR",
+            self.sink.ack.eq(0),
+            If(write_ptr + 1 == self.read_ptr_rx,
+                # if next one hasn't been read, overwrite the current buffer when new packet comes in
+                self.buffer_err.eq(1),
+            ).Else(
+                NextValue(write_ptr, write_ptr + 1),
+            ),
+            NextState("IDLE"),
+        )
+
 class CXP_Trig_Ack_Checker(Module, AutoCSR):
     def __init__(self):
         self.sink = stream.Endpoint(word_layout)
@@ -428,8 +477,10 @@ class CXP_Trig_Ack_Checker(Module, AutoCSR):
 
         self.submodules.fsm = fsm = FSM(reset_state="COPY")
 
+        self.submodules.voter = voter = Duplicate_Majority_Voter(self.sink.data, self.sink.k)
+
         fsm.act("COPY",
-            If((self.sink.stb & (self.sink.data == Replicate(KCode["io_ack"], 4)) & (self.sink.k == 0b1111)),
+            If((self.sink.stb & (voter.char == KCode["io_ack"]) & (voter.k == 1)),
                 # discard K28,6
                 self.sink.ack.eq(1),
                 NextState("CHECK_ACK")
@@ -443,8 +494,39 @@ class CXP_Trig_Ack_Checker(Module, AutoCSR):
                 NextState("COPY"),
                 # discard the word after K28,6
                 self.sink.ack.eq(1),
-                If(self.sink.data == Replicate(C(0x01, char_width), 4),
+                If((voter.char == 0x01) & (voter.k == 0),
                     self.ack.eq(1),
                 )
             )
         )
+
+@ResetInserter()
+@CEInserter()
+class CXPCRC32(Module):
+    # Section 9.2.2.2 (CXP-001-2021)
+    width = 32
+    polynom = 0x04C11DB7
+    seed = 2**width-1
+    check = 0x00000000
+    def __init__(self, data_width):
+        self.data = Signal(data_width)
+        self.value = Signal(self.width)
+        self.error = Signal()
+
+        # # #
+
+        self.submodules.engine = LiteEthMACCRCEngine(data_width, self.width, self.polynom)
+        reg = Signal(self.width, reset=self.seed)
+        self.sync += reg.eq(self.engine.next)
+        self.comb += [
+            self.engine.data.eq(self.data),
+            self.engine.last.eq(reg),
+
+            self.value.eq(reg[::-1]),
+            self.error.eq(self.engine.next != self.check)
+        ]
+
+# For verifying crc in stream data packet
+class CXPCRC32Checker(LiteEthMACCRCChecker):
+    def __init__(self, layout):
+        LiteEthMACCRCChecker.__init__(self, CXPCRC32, layout)

src/gateware/zc706.py

@@ -719,24 +719,25 @@ class CXP_FMC():
             cxp_csr_group.append(cxp_name)
             
 
-            mem_size = cxp_interface .get_mem_size()
-
             tx_mem_name = "cxp_tx" + str(i) + "_mem"
-            memory_address = self.axi2csr.register_port(cxp_interface.get_tx_port(), mem_size)
+            tx_mem_size = cxp_interface.get_tx_mem_size()
-            self.add_memory_region(tx_mem_name, self.mem_map["csr"] + memory_address, mem_size)
+            memory_address = self.axi2csr.register_port(cxp_interface.get_tx_port(), tx_mem_size)
+            self.add_memory_region(tx_mem_name, self.mem_map["csr"] + memory_address, tx_mem_size)
             cxp_tx_mem_group.append(tx_mem_name)
 
             rx_mem_name = "cxp_rx" + str(i) + "_mem"
-            memory_address = self.axi2csr.register_port(cxp_interface.get_rx_port(), mem_size)
+            rx_mem_size = cxp_interface.get_rx_mem_size()
-            self.add_memory_region(rx_mem_name, self.mem_map["csr"] + memory_address, mem_size)
+            memory_address = self.axi2csr.register_port(cxp_interface.get_rx_port(), rx_mem_size)
+            self.add_memory_region(rx_mem_name, self.mem_map["csr"] + memory_address, rx_mem_size)
             cxp_rx_mem_group.append(rx_mem_name)
 
 
             # DEBUG loopback tx memory
             loopback_mem_name = "cxp_loopback_tx" + str(i) + "_mem"
+            loopback_mem_size = cxp_interface.get_loopback_tx_mem_size()
             cxp_loopback_mem_group.append(loopback_mem_name)
-            memory_address = self.axi2csr.register_port(cxp_interface.get_loopback_tx_port(), mem_size)
+            memory_address = self.axi2csr.register_port(cxp_interface.get_loopback_tx_port(), loopback_mem_size)
-            self.add_memory_region(loopback_mem_name, self.mem_map["csr"] + memory_address, mem_size)
+            self.add_memory_region(loopback_mem_name, self.mem_map["csr"] + memory_address, loopback_mem_size)
 
         self.add_memory_group("cxp_tx_mem", cxp_tx_mem_group)
         self.add_memory_group("cxp_rx_mem", cxp_rx_mem_group)

src/libboard_artiq/Cargo.toml.tpl

@@ -23,6 +23,7 @@ core_io = { version = "0.1", features = ["collections"] }
 embedded-hal = "0.2"
 nb = "1.0"
 void = { version = "1", default-features = false }
+byteorder = { version = "1.3", default-features = false }
 
 io = { path = "../libio", features = ["byteorder"] }
 libboard_zynq = { path = "@@ZYNQ_RS@@/libboard_zynq" }

src/libboard_artiq/src/cxp_downconn.rs

@@ -43,16 +43,28 @@ pub fn loopback_testing(channel: usize, timer: &mut GlobalTimer, speed: CXP_SPEE
 
         cxp_proto::downconn_debug_send_trig_ack(channel);
 
+        const DATA_MAXSIZE: usize = 253;
+        let data_size = 4; // no. of bytes
+
+        let data: u32 = 0xDADA as u32;
+        let mut data_slice: [u8; DATA_MAXSIZE] = [0; DATA_MAXSIZE];
+        data_slice[..4].clone_from_slice(&data.to_be_bytes());
         cxp_proto::downconn_debug_send(
             channel,
-            &cxp_proto::Packet::CtrlRead {
+            &cxp_proto::UpConnPacket::Event {
-                addr: 0x00,
+                conn_id: 0x1234_5678_u32,
-                length: 0x04,
+                packet_tag: 0x69_u8,
+                length: data_size + 3,
+                event_size: data_size,
+                namespace: 0x02_u8,
+                event_id: 0x00_6969u16,
+                timestamp: 0x1234_5678u64,
+                data: data_slice,
             },
         )
         .expect("loopback gtx tx error");
 
-        timer.delay_us(200); // wait packet has arrive at RX async fifo
+        timer.delay_us(1000); // wait packet has arrive at RX async fifo
         (CXP[channel].downconn_tx_stb_write)(0);
 
         info!("trig ack = {}", (CXP[channel].downconn_trigger_ack_read)());
@@ -63,7 +75,9 @@ pub fn loopback_testing(channel: usize, timer: &mut GlobalTimer, speed: CXP_SPEE
         info!("test error = {}", (CXP[channel].downconn_test_error_read)());
         info!("packet type = {:#06X}", (CXP[channel].downconn_packet_type_read)());
 
-        cxp_proto::receive(channel);
+        cxp_proto::receive(channel).expect("loopback gtx rx error");
+        // cxp_proto::downconn_debug_mem_print(channel);
+
         // DEBUG: print loopback packets
         const LEN: usize = 20;
         let mut pak_arr: [u32; LEN] = [0; LEN];

src/libboard_artiq/src/cxp_proto.rs

@@ -1,21 +1,23 @@
 use core::slice;
 
-use core_io::{Error as IoError, Write};
+use byteorder::{ByteOrder, NetworkEndian};
-use crc::crc32;
+use core_io::{Error as IoError, Read, Write};
+use crc::crc32::checksum_ieee;
 use io::Cursor;
 use libboard_zynq::println;
 
-// TODO: fix the import
 use crate::{mem::mem::{CXP_LOOPBACK_MEM, CXP_RX_MEM, CXP_TX_MEM},
             pl::csr::CXP};
 
-const MAX_PACKET: usize = 128;
+const BUF_LEN: usize = 0x800;
-const DATA_MAXSIZE: usize = /*max size*/MAX_PACKET - /*Tag*/4 - /*Op code & length*/4 - /*addr*/4 - /*CRC*/4 ;
+const DATA_MAXSIZE: usize = 48;
-const MEM_LEN: usize = 0x200;
+const EV_MAXSIZE: usize = 253;
 
 #[derive(Debug)]
 pub enum Error {
     BufferError,
+    CorruptedPacket,
+    CtrlAckError(u8),
     LinkDown,
     UnknownPacket(u8),
 }
@@ -26,23 +28,303 @@ impl From<IoError> for Error {
     }
 }
 
-pub enum Packet {
+// Section 9.2.2.2 (CXP-001-2021)
+// Only Control packet need CRC32 appended in the end of the packet
+// CoaXpress use the polynomial of IEEE-802.3 (Ethernet) CRC but the checksum calculation is different
+fn get_cxp_crc(bytes: &[u8]) -> u32 {
+    (!checksum_ieee(bytes)).swap_bytes()
+}
+
+trait CxpRead {
+    fn read_u8(&mut self) -> Result<u8, Error>;
+
+    fn read_u16(&mut self) -> Result<u16, Error>;
+
+    fn read_u32(&mut self) -> Result<u32, Error>;
+
+    fn read_u64(&mut self) -> Result<u64, Error>;
+
+    fn read_exact_4x(&mut self, buf: &mut [u8]) -> Result<(), Error>;
+
+    fn read_4x_u8(&mut self) -> Result<u8, Error>;
+
+    fn read_4x_u16(&mut self) -> Result<u16, Error>;
+
+    fn read_4x_u32(&mut self) -> Result<u32, Error>;
+
+    fn read_4x_u64(&mut self) -> Result<u64, Error>;
+}
+impl<Cursor: Read> CxpRead for Cursor {
+    fn read_u8(&mut self) -> Result<u8, Error> {
+        let mut bytes = [0; 1];
+        self.read_exact(&mut bytes)?;
+        Ok(bytes[0])
+    }
+
+    fn read_u16(&mut self) -> Result<u16, Error> {
+        let mut bytes = [0; 2];
+        self.read_exact(&mut bytes)?;
+        Ok(NetworkEndian::read_u16(&bytes))
+    }
+
+    fn read_u32(&mut self) -> Result<u32, Error> {
+        let mut bytes = [0; 4];
+        self.read_exact(&mut bytes)?;
+        Ok(NetworkEndian::read_u32(&bytes))
+    }
+
+    fn read_u64(&mut self) -> Result<u64, Error> {
+        let mut bytes = [0; 8];
+        self.read_exact(&mut bytes)?;
+        Ok(NetworkEndian::read_u64(&bytes))
+    }
+
+    fn read_exact_4x(&mut self, buf: &mut [u8]) -> Result<(), Error> {
+        for byte in buf {
+            // Section 9.2.2.1 (CXP-001-2021)
+            // decoder should immune to single bit errors when handling 4x duplicated characters
+            let a = self.read_u8()?;
+            let b = self.read_u8()?;
+            let c = self.read_u8()?;
+            let d = self.read_u8()?;
+            // vote and return majority
+            *byte = a & b & c | a & b & d | a & c & d | b & c & d;
+        }
+        Ok(())
+    }
+
+    fn read_4x_u8(&mut self) -> Result<u8, Error> {
+        let mut bytes = [0; 1];
+        self.read_exact_4x(&mut bytes)?;
+        Ok(bytes[0])
+    }
+
+    fn read_4x_u16(&mut self) -> Result<u16, Error> {
+        let mut bytes = [0; 2];
+        self.read_exact_4x(&mut bytes)?;
+        Ok(NetworkEndian::read_u16(&bytes))
+    }
+
+    fn read_4x_u32(&mut self) -> Result<u32, Error> {
+        let mut bytes = [0; 4];
+        self.read_exact_4x(&mut bytes)?;
+        Ok(NetworkEndian::read_u32(&bytes))
+    }
+
+    fn read_4x_u64(&mut self) -> Result<u64, Error> {
+        let mut bytes = [0; 6];
+        self.read_exact_4x(&mut bytes)?;
+        Ok(NetworkEndian::read_u64(&bytes))
+    }
+}
+
+#[derive(Debug)]
+pub enum NameSpace {
+    GenICam,
+    DeviceSpecific,
+}
+
+#[derive(Debug)]
+pub enum DownConnPacket {
+    CtrlReply {
+        tag: Option<u8>,
+        length: u32,
+        data: [u8; DATA_MAXSIZE],
+    },
+    CtrlDelay {
+        tag: Option<u8>,
+        length: u32,
+        time: [u8; DATA_MAXSIZE],
+    },
+    CtrlAck {
+        tag: Option<u8>,
+    },
+    Event {
+        conn_id: u32,
+        packet_tag: u8,
+        length: u16,
+        ev_size: u16,
+        namespace: NameSpace,
+        event_id: u16,
+        timestamp: u64,
+        ev: [u8; EV_MAXSIZE],
+    },
+}
+
+impl DownConnPacket {
+    pub fn read_from(reader: &mut Cursor<&mut [u8]>, packet_type: u8) -> Result<Self, Error> {
+        match packet_type {
+            0x03 => DownConnPacket::get_ctrl_packet(reader, false),
+            0x06 => DownConnPacket::get_ctrl_packet(reader, true),
+            0x07 => DownConnPacket::get_event_packet(reader),
+            _ => Err(Error::UnknownPacket(packet_type)),
+        }
+    }
+
+    fn get_ctrl_packet(reader: &mut Cursor<&mut [u8]>, with_tag: bool) -> Result<Self, Error> {
+        let mut tag: Option<u8> = None;
+        if with_tag {
+            tag = Some(reader.read_4x_u8()?);
+        }
+
+        let ackcode = reader.read_4x_u8()?;
+
+        match ackcode {
+            0x00 | 0x04 => {
+                let length = reader.read_u32()?;
+                let mut data: [u8; DATA_MAXSIZE] = [0; DATA_MAXSIZE];
+                reader.read(&mut data[0..length as usize])?;
+
+                let checksum = get_cxp_crc(&reader.get_ref()[0..reader.position()]);
+                if reader.read_u32()? != checksum {
+                    return Err(Error::CorruptedPacket);
+                }
+
+                if ackcode == 0x00 {
+                    return Ok(DownConnPacket::CtrlReply { tag, length, data });
+                } else {
+                    return Ok(DownConnPacket::CtrlDelay {
+                        tag,
+                        length,
+                        time: data,
+                    });
+                }
+            }
+            0x01 => return Ok(DownConnPacket::CtrlAck { tag }),
+            _ => return Err(Error::CtrlAckError(ackcode)),
+        }
+    }
+
+    fn get_event_packet(reader: &mut Cursor<&mut [u8]>) -> Result<Self, Error> {
+        let conn_id = reader.read_4x_u32()?;
+        let packet_tag = reader.read_4x_u8()?;
+        let length = reader.read_4x_u16()?;
+
+        let ev_size = reader.read_u16()?;
+        if ev_size + 3 != length {
+            println!("length mismatch");
+            return Err(Error::CorruptedPacket);
+        }
+
+        let mut bytes = [0; 2];
+        reader.read_exact(&mut bytes)?;
+        let namespace_bits = (bytes[0] & 0xC0) >> 6;
+        let namespace = match namespace_bits {
+            0 => NameSpace::GenICam,
+            2 => NameSpace::DeviceSpecific,
+            _ => {
+                println!("namespace = {} error", namespace_bits);
+                return Err(Error::CorruptedPacket);
+            }
+        };
+
+        let event_id = (bytes[0] & 0xF) as u16 | (bytes[1] as u16);
+
+        let timestamp = reader.read_u64()?;
+
+        let mut ev: [u8; EV_MAXSIZE] = [0; EV_MAXSIZE];
+        reader.read(&mut ev[0..ev_size as usize])?;
+
+        let checksum = get_cxp_crc(&reader.get_ref()[0..reader.position()]);
+        if reader.read_u32()? != checksum {
+            println!("crc error");
+            return Err(Error::CorruptedPacket);
+        }
+
+        Ok(DownConnPacket::Event {
+            conn_id,
+            packet_tag,
+            length,
+            ev_size,
+            namespace,
+            event_id,
+            timestamp,
+            ev,
+        })
+    }
+}
+
+pub fn receive(channel: usize) -> Result<Option<DownConnPacket>, Error> {
+    unsafe {
+        if (CXP[channel].downconn_pending_packet_read)() == 1 {
+            let read_buffer_ptr = (CXP[channel].downconn_read_ptr_read)() as usize;
+            println!("buffer ptr = {}", read_buffer_ptr);
+            let ptr = (CXP_RX_MEM[channel].base + read_buffer_ptr * BUF_LEN) as *mut u8;
+
+            let mut reader = Cursor::new(slice::from_raw_parts_mut(ptr, BUF_LEN));
+            let packet_type = (CXP[channel].downconn_packet_type_read)();
+
+            let packet = DownConnPacket::read_from(&mut reader, packet_type);
+            println!("{:X?}", packet);
+
+            (CXP[channel].downconn_pending_packet_write)(1);
+            Ok(Some(packet?))
+        } else {
+            Ok(None)
+        }
+    }
+}
+
+trait CxpWrite {
+    fn write_all_4x(&mut self, buf: &[u8]) -> Result<(), Error>;
+
+    fn write_4x_u8(&mut self, value: u8) -> Result<(), Error>;
+
+    fn write_4x_u16(&mut self, value: u16) -> Result<(), Error>;
+
+    fn write_4x_u32(&mut self, value: u32) -> Result<(), Error>;
+
+    fn write_4x_u64(&mut self, value: u64) -> Result<(), Error>;
+
+    fn write_u32(&mut self, value: u32) -> Result<(), Error>;
+}
+impl<Cursor: Write> CxpWrite for Cursor {
+    fn write_all_4x(&mut self, buf: &[u8]) -> Result<(), Error> {
+        for byte in buf {
+            self.write_all(&[*byte; 4])?;
+        }
+        Ok(())
+    }
+
+    fn write_4x_u8(&mut self, value: u8) -> Result<(), Error> {
+        self.write_all_4x(&[value])
+    }
+
+    fn write_4x_u16(&mut self, value: u16) -> Result<(), Error> {
+        let mut bytes = [0; 2];
+        NetworkEndian::write_u16(&mut bytes, value);
+        self.write_all_4x(&bytes)
+    }
+
+    fn write_4x_u32(&mut self, value: u32) -> Result<(), Error> {
+        let mut bytes = [0; 4];
+        NetworkEndian::write_u32(&mut bytes, value);
+        self.write_all_4x(&bytes)
+    }
+
+    fn write_4x_u64(&mut self, value: u64) -> Result<(), Error> {
+        let mut bytes = [0; 6];
+        NetworkEndian::write_u64(&mut bytes, value);
+        self.write_all_4x(&bytes)
+    }
+
+    fn write_u32(&mut self, value: u32) -> Result<(), Error> {
+        let mut bytes = [0; 4];
+        NetworkEndian::write_u32(&mut bytes, value);
+        self.write_all(&bytes)?;
+        Ok(())
+    }
+}
+
+#[derive(Debug)]
+pub enum UpConnPacket {
     CtrlRead {
+        tag: Option<u8>,
         addr: u32,
         length: u8,
     },
     CtrlWrite {
-        addr: u32,
+        tag: Option<u8>,
-        length: u8,
-        data: [u8; DATA_MAXSIZE],
-    }, // max register size is 8 bytes
-    CtrlReadWithTag {
-        tag: u8,
-        addr: u32,
-        length: u8,
-    },
-    CtrlWriteWithTag {
-        tag: u8,
         addr: u32,
         length: u8,
         data: [u8; DATA_MAXSIZE],
@@ -51,58 +333,113 @@ pub enum Packet {
         packet_tag: u8,
     },
     TestPacket,
+
+    // DEBUG: Loopback message
+    CtrlAckLoopback {
+        ackcode: u8,
+        length: u8,
+        data: [u8; DATA_MAXSIZE],
+    },
+    Event {
+        conn_id: u32,
+        packet_tag: u8,
+        length: u16,
+        event_size: u16,
+        namespace: u8,
+        event_id: u16,
+        timestamp: u64,
+        data: [u8; 253],
+    },
 }
 
-impl Packet {
+impl UpConnPacket {
     pub fn write_to(&self, writer: &mut Cursor<&mut [u8]>) -> Result<(), Error> {
-        // CoaXpress use big endian
         match *self {
-            Packet::CtrlRead { addr, length } => {
+            UpConnPacket::CtrlRead { tag, addr, length } => {
-                writer.write(&[0x02; 4])?;
+                match tag {
-                writer.write(&[0x00, 0x00, 0x00, length])?;
+                    Some(t) => {
-                writer.write(&addr.to_be_bytes())?;
+                        writer.write_4x_u8(0x05)?;
+                        writer.write_4x_u8(t)?;
+                    }
+                    None => {
+                        writer.write_4x_u8(0x02)?;
+                    }
+                }
+                writer.write_all(&[0x00, 0x00, 0x00, length])?;
+                writer.write_u32(addr)?;
+
+                // Section 9.6.2 (CXP-001-2021)
+                // only bytes after the first 4 are used in calculating the checksum
+                let checksum = get_cxp_crc(&writer.get_ref()[4..writer.position()]);
+                writer.write_u32(checksum)?;
             }
-            Packet::CtrlWrite { addr, length, data } => {
+            UpConnPacket::CtrlWrite {
-                writer.write(&[0x02; 4])?;
-                writer.write(&[0x01, 0x00, 0x00, length])?;
-                writer.write(&addr.to_be_bytes())?;
-                writer.write(&data[0..length as usize])?;
-            }
-            Packet::CtrlReadWithTag { tag, addr, length } => {
-                writer.write(&[0x05; 4])?;
-                writer.write(&[tag; 4])?;
-                writer.write(&[0x00, 0x00, 0x00, length])?;
-                writer.write(&addr.to_be_bytes())?;
-            }
-            Packet::CtrlWriteWithTag {
                 tag,
                 addr,
                 length,
                 data,
             } => {
-                writer.write(&[0x05; 4])?;
+                match tag {
-                writer.write(&[tag; 4])?;
+                    Some(t) => {
-                writer.write(&[0x01, 0x00, 0x00, length])?;
+                        writer.write_4x_u8(0x05)?;
-                writer.write(&addr.to_be_bytes())?;
+                        writer.write_4x_u8(t)?;
-                writer.write(&data[0..length as usize])?;
+                    }
+                    None => {
+                        writer.write_4x_u8(0x02)?;
+                    }
+                }
+                writer.write_all(&[0x01, 0x00, 0x00, length])?;
+                writer.write_u32(addr)?;
+                writer.write_all(&data[0..length as usize])?;
+
+                // Section 9.6.2 (CXP-001-2021)
+                // only bytes after the first 4 are used in calculating the checksum
+                let checksum = get_cxp_crc(&writer.get_ref()[4..writer.position()]);
+                writer.write_u32(checksum)?;
             }
-            Packet::EventAck { packet_tag } => {
+            UpConnPacket::EventAck { packet_tag } => {
-                writer.write(&[0x08; 4])?;
+                writer.write_4x_u8(0x08)?;
-                writer.write(&[packet_tag; 4])?;
+                writer.write_4x_u8(packet_tag)?;
             }
-            _ => {}
+            // DEBUG: Loopback message
-        }
+            UpConnPacket::CtrlAckLoopback { ackcode, length, data } => {
-        // Section 9.2.2.2 (CXP-001-2021)
+                writer.write_4x_u8(0x03)?;
-        // Only Control packet need CRC32 appended in the end of the packet
+                writer.write_4x_u8(ackcode)?;
-        // CoaXpress use the polynomial of IEEE-802.3 (Ethernet) CRC but the checksum calculation is different
+
-        // Also, the calculation does not include the first 4 bytes of packet_type
+                if ackcode == 0x00 || ackcode == 0x04 {
-        match *self {
+                    writer.write_all(&[0x00, 0x00, 0x00, length])?;
-            Packet::CtrlRead { .. }
+                    writer.write_all(&data[0..length as usize])?;
-            | Packet::CtrlWrite { .. }
+                }
-            | Packet::CtrlReadWithTag { .. }
+
-            | Packet::CtrlWriteWithTag { .. } => {
+                let checksum = get_cxp_crc(&writer.get_ref()[4..writer.position()]);
-                let checksum = crc32::checksum_ieee(&writer.get_ref()[4..writer.position()]);
+                writer.write_u32(checksum)?;
-                writer.write(&(!checksum).to_le_bytes())?;
+            }
+            UpConnPacket::Event {
+                conn_id,
+                packet_tag,
+                length,
+                event_size,
+                namespace,
+                event_id,
+                timestamp,
+                data,
+            } => {
+                // event packet header
+                writer.write_4x_u8(0x07)?;
+                writer.write_4x_u32(conn_id)?;
+                writer.write_4x_u8(packet_tag)?;
+                writer.write_4x_u16(length)?;
+
+                // event message
+                let ev_size = event_size.to_be_bytes();
+                let p2: u8 = ((namespace & 0b11) << 6) | ((event_id & 0xF00) >> 8) as u8;
+                let p3: u8 = (event_id & 0xFF) as u8;
+                writer.write_all(&[ev_size[0], ev_size[1], p2, p3])?;
+                writer.write_all(&timestamp.to_be_bytes())?;
+                writer.write_all(&data[0..event_size as usize])?;
+
+                let checksum = get_cxp_crc(&writer.get_ref()[4..writer.position()]);
+                writer.write_u32(checksum)?;
             }
             _ => {}
         }
@@ -110,38 +447,27 @@ impl Packet {
     }
 }
 
-pub fn receive(channel: usize) -> Result<(), Error> {
+pub fn send(channel: usize, packet: &UpConnPacket) -> Result<(), Error> {
-    unsafe {
-        // let ptr = CXP_LOOPBACK_MEM[0].base as *mut u32;
-        let ptr = CXP_RX_MEM[0].base as *mut u32;
-        // let mut reader = Cursor::new(slice::from_raw_parts_mut(ptr as *mut u8, MEM_LEN));
-
-        print_packet(slice::from_raw_parts_mut(ptr as *mut u8, MEM_LEN));
-    }
-    Ok(())
-}
-
-pub fn send(channel: usize, packet: &Packet) -> Result<(), Error> {
     if unsafe { (CXP[channel].upconn_tx_enable_read)() } == 0 {
         Err(Error::LinkDown)?
     }
 
     match *packet {
-        Packet::TestPacket => send_test_packet(channel),
+        UpConnPacket::TestPacket => send_test_packet(channel),
         _ => send_data_packet(channel, packet),
     }
 }
 
-fn send_data_packet(channel: usize, packet: &Packet) -> Result<(), Error> {
+fn send_data_packet(channel: usize, packet: &UpConnPacket) -> Result<(), Error> {
     unsafe {
         // TODO: put this in mem group
         while (CXP[channel].upconn_command_tx_read)() == 1 {}
         let ptr = CXP_TX_MEM[0].base as *mut u32;
-        let mut writer = Cursor::new(slice::from_raw_parts_mut(ptr as *mut u8, MEM_LEN));
+        let mut writer = Cursor::new(slice::from_raw_parts_mut(ptr as *mut u8, BUF_LEN));
 
         packet.write_to(&mut writer)?;
 
-        (CXP[channel].upconn_command_tx_word_len_write)(writer.position() as u8 / 4);
+        (CXP[channel].upconn_command_tx_word_len_write)(writer.position() as u16 / 4);
         (CXP[channel].upconn_command_tx_write)(1);
     }
 
@@ -161,43 +487,56 @@ fn send_test_packet(channel: usize) -> Result<(), Error> {
     Ok(())
 }
 
-pub fn write_u32(channel: usize, addr: u32, data: u32) -> Result<(), Error> {
+//
-    let mut data_slice: [u8; DATA_MAXSIZE] = [0; DATA_MAXSIZE];
-    data_slice[..4].clone_from_slice(&data.to_be_bytes());
-    send(
-        channel,
-        &Packet::CtrlWrite {
-            addr,
-            length: 4,
-            data: data_slice,
-        },
-    )?;
-
-    Ok(())
-}
-
-pub fn read_u32(channel: usize, addr: u32) -> Result<(), Error> {
-    send(channel, &Packet::CtrlRead { addr, length: 4 })?;
-
-    Ok(())
-}
-
-pub fn write_u64(channel: usize, addr: u32, data: u64) -> Result<(), Error> {
-    let mut data_slice: [u8; DATA_MAXSIZE] = [0; DATA_MAXSIZE];
-    data_slice[..8].clone_from_slice(&data.to_be_bytes());
-    send(
-        channel,
-        &Packet::CtrlWrite {
-            addr,
-            length: 8,
-            data: data_slice,
-        },
-    )?;
-
-    Ok(())
-}
-
 // DEBUG: use only
+//
+//
+//
+// pub fn write_u32(channel: usize, addr: u32, data: u32) -> Result<(), Error> {
+//     let mut data_slice: [u8; DATA_MAXSIZE] = [0; DATA_MAXSIZE];
+//     data_slice[..4].clone_from_slice(&data.to_be_bytes());
+//     send(
+//         channel,
+//         &UpConnPacket::CtrlWrite {
+//             tag: None,
+//             addr,
+//             length: 4,
+//             data: data_slice,
+//         },
+//     )?;
+
+//     Ok(())
+// }
+
+// pub fn read_u32(channel: usize, addr: u32) -> Result<(), Error> {
+//     send(
+//         channel,
+//         &UpConnPacket::CtrlRead {
+//             tag: None,
+//             addr,
+//             length: 4,
+//         },
+//     )?;
+
+//     Ok(())
+// }
+
+// pub fn write_u64(channel: usize, addr: u32, data: u64) -> Result<(), Error> {
+//     let mut data_slice: [u8; DATA_MAXSIZE] = [0; DATA_MAXSIZE];
+//     data_slice[..8].clone_from_slice(&data.to_be_bytes());
+//     send(
+//         channel,
+//         &UpConnPacket::CtrlWrite {
+//             tag: None,
+//             addr,
+//             length: 8,
+//             data: data_slice,
+//         },
+//     )?;
+
+//     Ok(())
+// }
+
 pub fn print_packet(pak: &[u8]) {
     println!("pak = [");
     for i in 0..(pak.len() / 4) {
@@ -232,22 +571,29 @@ pub fn print_packetu32(pak: &[u32], k: &[u8]) {
     println!("============================================");
 }
 
-pub fn downconn_debug_send(channel: usize, packet: &Packet) -> Result<(), Error> {
+pub fn downconn_debug_send(channel: usize, packet: &UpConnPacket) -> Result<(), Error> {
     unsafe {
-        // TODO: put this in mem group
         while (CXP[channel].downconn_command_tx_read)() == 1 {}
         let ptr = CXP_LOOPBACK_MEM[0].base as *mut u32;
-        let mut writer = Cursor::new(slice::from_raw_parts_mut(ptr as *mut u8, MEM_LEN));
+        let mut writer = Cursor::new(slice::from_raw_parts_mut(ptr as *mut u8, BUF_LEN));
 
         packet.write_to(&mut writer)?;
 
-        (CXP[channel].downconn_command_tx_word_len_write)(writer.position() as u8 / 4);
+        (CXP[channel].downconn_command_tx_word_len_write)(writer.position() as u16 / 4);
         (CXP[channel].downconn_command_tx_write)(1);
     }
 
     Ok(())
 }
 
+pub fn downconn_debug_mem_print(channel: usize) {
+    unsafe {
+        let ptr = CXP_RX_MEM[channel].base as *mut u8;
+        let arr = slice::from_raw_parts_mut(ptr, BUF_LEN * 4);
+        print_packet(arr);
+    }
+}
+
 pub fn downconn_debug_send_trig_ack(channel: usize) {
     unsafe {
         (CXP[channel].downconn_ack_write)(1);

src/libboard_artiq/src/cxp_upconn.rs

@@ -15,7 +15,7 @@ pub fn tx_test(channel: usize, timer: &mut GlobalTimer) {
 
         (CXP[channel].upconn_tx_enable_write)(1);
         timer.delay_us(2); // send one word
-        cxp_proto::read_u32(channel, 0x00).expect("Cannot Write CoaXpress Register");
+        // cxp_proto::read_u32(channel, 0x00).expect("Cannot Write CoaXpress Register");
         // cxp_proto::write_u64(channel, 0x00, 0x01);
         // cxp_proto::send(channel, &cxp_proto::Packet::EventAck { packet_tag: 0x04 }).expect("Cannot send CoaXpress packet");
         // cxp_proto::send(channel, &cxp_proto::Packet::TestPacket).expect("Cannot send CoaXpress packet");

src/libboard_artiq/src/lib.rs

@@ -3,6 +3,7 @@
 #![feature(naked_functions)]
 #![feature(asm)]
 
+extern crate byteorder;
 extern crate core_io;
 extern crate crc;
 extern crate embedded_hal;