Add new packet formats for LWA Stations

configure

@@ -742,6 +742,7 @@ RDMA_MAXMEM
 HAVE_RDMA
 VERBS_SEND_PACING
 VERBS_SEND_NPKTBUF
+VERBS_WCBATCH
 VERBS_NPKTBUF
 HAVE_VERBS
 HAVE_VMA
@@ -861,6 +862,7 @@ enable_hwloc
 enable_vma
 enable_verbs
 with_rx_buffer_size
+with_verbs_wcbatch
 with_tx_buffer_size
 enable_rdma
 with_rdma_max_mem
@@ -1584,7 +1586,9 @@ Optional Packages:
                           compiler's sysroot if not specified).
   --with-ctags=[PATH]   absolute path to ctags executable
   --with-rx-buffer-size=N default Infiniband verbs receive buffer size in
-                          packets (default=8192)
+                          packets (default=32768)
+  --with-verbs-wcbatch=N  default Infiniband verbs completion queue batch size
+                          (default=64)
   --with-tx-buffer-size=N default Infiniband verbs send buffer size in packets
                           (default=512)
   --with-rdma-max-mem=N   maximum RDMA buffer size in bytes
@@ -21549,7 +21553,7 @@ if test ${with_rx_buffer_size+y}
 then :
   withval=$with_rx_buffer_size;
 else case e in #(
-  e) with_rx_buffer_size=8192 ;;
+  e) with_rx_buffer_size=32768 ;;
 esac
 fi
 
@@ -21562,6 +21566,24 @@ then :
 fi
 
 
+# Check whether --with-verbs-wcbatch was given.
+if test ${with_verbs_wcbatch+y}
+then :
+  withval=$with_verbs_wcbatch;
+else case e in #(
+  e) with_verbs_wcbatch=64 ;;
+esac
+fi
+
+ VERBS_WCBATCH=$with_verbs_wcbatch
+
+ if test x$HAVE_VERBS = x0
+then :
+  VERBS_WCBATCH=0
+
+fi
+
+
 # Check whether --with-tx-buffer-size was given.
 if test ${with_tx_buffer_size+y}
 then :

configure.ac

@@ -141,13 +141,22 @@ AS_IF([test x$enable_verbs != xno],
 
  AC_ARG_WITH([rx-buffer-size],
             [AS_HELP_STRING([--with-rx-buffer-size=N],
-                            [default Infiniband verbs receive buffer size in packets (default=8192)])],
+                            [default Infiniband verbs receive buffer size in packets (default=32768)])],
             [],
-            [with_rx_buffer_size=8192])
+            [with_rx_buffer_size=32768])
  AC_SUBST([VERBS_NPKTBUF], [$with_rx_buffer_size])
  AS_IF([test x$HAVE_VERBS = x0],
        [AC_SUBST([VERBS_NPKTBUF], [0])])
 
+ AC_ARG_WITH([verbs-wcbatch],
+            [AS_HELP_STRING([--with-verbs-wcbatch=N],
+                            [default Infiniband verbs completion queue batch size (default=64)])],
+            [],
+            [with_verbs_wcbatch=64])
+ AC_SUBST([VERBS_WCBATCH], [$with_verbs_wcbatch])
+ AS_IF([test x$HAVE_VERBS = x0],
+       [AC_SUBST([VERBS_WCBATCH], [0])])
+
  AC_ARG_WITH([tx-buffer-size],
             [AS_HELP_STRING([--with-tx-buffer-size=N],
                             [default Infiniband verbs send buffer size in packets (default=512)])],

python/bifrost/packet_capture.py

@@ -60,6 +60,10 @@ def set_snap2(self, fnc: _bf.BFpacketcapture_snap2_sequence_callback):
         self._ref_cache['snap2'] = _bf.BFpacketcapture_snap2_sequence_callback(fnc)
         _check(_bf.bfPacketCaptureCallbackSetSNAP2(
                self.obj, self._ref_cache['snap2']))
+    def set_zcu102(self, fnc: _bf.BFpacketcapture_zcu102_sequence_callback):
+        self._ref_cache['zcu102'] = _bf.BFpacketcapture_zcu102_sequence_callback(fnc)
+        _check(_bf.bfPacketCaptureCallbackSetZCU102(
+               self.obj, self._ref_cache['zcu102']))
     def set_ibeam(self, fnc: _bf.BFpacketcapture_ibeam_sequence_callback):
         self._ref_cache['ibeam'] = _bf.BFpacketcapture_ibeam_sequence_callback(fnc)
         _check(_bf.bfPacketCaptureCallbackSetIBeam(

src/bifrost/config.h.in

@@ -66,6 +66,7 @@ extern "C" {
 #define BF_VERBS_ENABLED @HAVE_VERBS@
 #define BF_VERBS_NPKTBUF @VERBS_NPKTBUF@
 #define BF_VERBS_SEND_NPKTBUF @VERBS_SEND_NPKTBUF@
+#define BF_VERBS_WCBATCH @VERBS_WCBATCH@
 #define BF_VERBS_SEND_PACING @VERBS_SEND_PACING@
 #define BF_RDMA_ENABLED @HAVE_RDMA@
 #define BF_RDMA_MAXMEM @RDMA_MAXMEM@

src/bifrost/packet_capture.h

@@ -44,6 +44,8 @@ typedef int (*BFpacketcapture_chips_sequence_callback)(BFoffset, int, int, int,
                                                        BFoffset*, void const**, size_t*);
 typedef int (*BFpacketcapture_snap2_sequence_callback)(BFoffset, int, int, int,
                                                        BFoffset*, void const**, size_t*);
+typedef int (*BFpacketcapture_zcu102_sequence_callback)(BFoffset, int, int, int,
+                                                        BFoffset*, void const**, size_t*);
 typedef int (*BFpacketcapture_ibeam_sequence_callback)(BFoffset, int, int, int,
                                                        BFoffset*, void const**, size_t*);
 typedef int (*BFpacketcapture_pbeam_sequence_callback)(BFoffset, BFoffset, int, int, int,
@@ -69,6 +71,8 @@ BFstatus bfPacketCaptureCallbackSetCHIPS(BFpacketcapture_callback obj,
                                          BFpacketcapture_chips_sequence_callback callback);
 BFstatus bfPacketCaptureCallbackSetSNAP2(BFpacketcapture_callback obj,
                                          BFpacketcapture_snap2_sequence_callback callback);
+BFstatus bfPacketCaptureCallbackSetZCU102(BFpacketcapture_callback obj,
+                                          BFpacketcapture_zcu102_sequence_callback callback);
 BFstatus bfPacketCaptureCallbackSetIBeam(BFpacketcapture_callback obj,
                                          BFpacketcapture_ibeam_sequence_callback callback);
 BFstatus bfPacketCaptureCallbackSetPBeam(BFpacketcapture_callback obj,

src/formats/cor.hpp

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2019-2021, The Bifrost Authors. All rights reserved.
+ * Copyright (c) 2019-2025, The Bifrost Authors. All rights reserved.
  *
  * Redistribution and use in source and binary forms, with or without
  * modification, are permitted provided that the following conditions
@@ -72,7 +72,7 @@ class CORDecoder : virtual public PacketDecoder {
 	    pkt->sync         = pkt_hdr->sync_word;
 	    pkt->time_tag     = be64toh(pkt_hdr->time_tag);
 	    pkt->decimation   = be32toh(pkt_hdr->navg);
-	    pkt->seq          = pkt->time_tag / 196000000 / (pkt->decimation / 100);
+	    pkt->seq          = pkt->time_tag / pkt->decimation;
 	    pkt->nsrc         = _nsrc;
 	    pkt->src          = (stand0*(2*(nstand-1)+1-stand0)/2 + stand1 + 1 - _src0)*nserver \
 	                        + (server - 1);

src/formats/drx.hpp

@@ -134,7 +134,6 @@ class DRXProcessor : virtual public PacketProcessor {
 			    *(aligned_data + src) = 0;
 			    aligned_data += nsrc;
 		    }
-		    aligned_data += t*4096*nsrc;
 	    }
     }
 };

src/formats/drx8.hpp

@@ -134,7 +134,6 @@ class DRX8Processor : virtual public PacketProcessor {
 			    *(aligned_data + src) = 0;
 			    aligned_data += nsrc;
 		    }
-		    aligned_data += t*4096*nsrc;
 	    }
     }
 };

src/formats/formats.hpp

@@ -40,3 +40,5 @@
 #include "pbeam.hpp"
 #include "simple.hpp"
 #include "vbeam.hpp"
+#include "zcu102.hpp"
+#include "tbx.hpp"

src/formats/ibeam.hpp

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2019, The Bifrost Authors. All rights reserved.
+ * Copyright (c) 2019-2025, The Bifrost Authors. All rights reserved.
  *
  * Redistribution and use in source and binary forms, with or without
  * modification, are permitted provided that the following conditions
@@ -33,11 +33,11 @@
 //#include <immintrin.h> // SSE
 
 struct __attribute__((packed)) ibeam_hdr_type {
-	uint8_t  server;   // Note: 1-based
-	uint8_t  gbe;      // (AKA tuning)
-	uint8_t  nchan;    // 109
-	uint8_t  nbeam;    // 2
-	uint8_t  nserver;  // 6
+	uint16_t server;   // Note: 1-based
+	uint16_t gbe;      // (AKA tuning)
+	uint16_t nchan;    // 109
+	uint16_t nbeam;    // 2
+	uint16_t nserver;  // 6
 	// Note: Big endian
 	uint16_t chan0;    // First chan in packet
 	uint64_t seq;      // Note: 1-based
@@ -67,9 +67,9 @@ class IBeamDecoder: virtual public PacketDecoder {
 	    pkt->seq   = be64toh(pkt_hdr->seq)  - 1;
 	    //pkt->nsrc  =         pkt_hdr->nserver;
 	    pkt->nsrc  =         _nsrc;
-	    pkt->src   =        (pkt_hdr->server - 1) - _src0;
-	    pkt->beam  =         pkt_hdr->nbeam;
-	    pkt->nchan =         pkt_hdr->nchan;
+	    pkt->src   =  (ntohs(pkt_hdr->server) - 1) - _src0;
+	    pkt->beam  =   ntohs(pkt_hdr->nbeam);
+	    pkt->nchan =   ntohs(pkt_hdr->nchan);
 	    pkt->chan0 =   ntohs(pkt_hdr->chan0) - pkt->nchan * pkt->src;
 	    pkt->payload_size = pld_size;
 	    pkt->payload_ptr  = pkt_pld;
@@ -164,11 +164,11 @@ class IBeamHeaderFiller : virtual public PacketHeaderFiller {
         ibeam_hdr_type* header = reinterpret_cast<ibeam_hdr_type*>(hdr);
         memset(header, 0, sizeof(ibeam_hdr_type));
 
-        header->server   = hdr_base->src + 1;
-        header->gbe      = hdr_base->tuning;
-        header->nchan    = hdr_base->nchan;
-        header->nbeam    = _nbeam;
-        header->nserver  = hdr_base->nsrc;
+        header->server   = htons(hdr_base->src + 1);
+        header->gbe      = htons(hdr_base->tuning);
+        header->nchan    = htons(hdr_base->nchan);
+        header->nbeam    = htons(_nbeam);
+        header->nserver  = htons(hdr_base->nsrc);
         header->chan0    = htons(hdr_base->chan0);
         header->seq      = htobe64(hdr_base->seq);
     }

src/formats/snap2.hpp

@@ -156,24 +156,21 @@ class SNAP2Processor : virtual public PacketProcessor {
             int c=0;
 #if defined BF_AVX_ENABLED && BF_AVX_ENABLED
             __m256i *dest_p;
-            __m256i vecbuf[2];
-            uint64_t *in64 = (uint64_t *)in;
+            __m256i *src_p = (__m256i *)in;
             dest_p = (__m256i *)(out + (words_per_chan_out * (pkt_chan)) + pol_offset_out);
 #endif
             //if((pol_offset_out == 0) && (pkt_chan==0) && ((pkt->seq % 120)==0) ){
             //   fprintf(stderr, "nsrc: %d seq: %d, dest_p: %p obuf idx %d, obuf offset %lu, nseq_per_obuf %d, seq0 %d, nbuf: %d\n", pkt->nsrc, pkt->seq, dest_p, obuf_idx, obuf_offset, nseq_per_obuf, seq0, nbuf);
             //}
             for(c=0; c<pkt->nchan; c++) {
 #if defined BF_AVX_ENABLED && BF_AVX_ENABLED
-               vecbuf[0] = _mm256_set_epi64x(in64[3], in64[2], in64[1], in64[0]);
-               vecbuf[1] = _mm256_set_epi64x(in64[7], in64[6], in64[5], in64[4]);
-               _mm256_stream_si256(dest_p, vecbuf[0]);
-               _mm256_stream_si256(dest_p+1,   vecbuf[1]);
-               in64 += 8;
+               _mm256_stream_si256(dest_p,   _mm256_loadu_si256(src_p));
+               _mm256_stream_si256(dest_p+1, _mm256_loadu_si256(src_p+1));
+               src_p += 2;
                dest_p += words_per_chan_out;
 #else
-               ::memcpy(&out[pkt->src + pkt->nsrc*c],
-                        &in[c], sizeof(otype));
+               ::memcpy(out + (words_per_chan_out * (pkt_chan + c)) + pol_offset_out,
+                        (uint8_t*)in + c * pkt->npol, pkt->npol);
 #endif
             }
     }
@@ -184,13 +181,21 @@ class SNAP2Processor : virtual public PacketProcessor {
                                  int      nchan,
                                  int      nseq) {
             typedef aligned256_type otype;
-            otype* __restrict__ aligned_data = (otype*)data;
-            for( int t=0; t<nseq; ++t ) {
-                   for( int c=0; c<nchan; ++c ) {
-                           ::memset(&aligned_data[src + nsrc*(c + nchan*t)],
-                                    0, sizeof(otype));
-                   }
+            otype* __restrict__ ptr = (otype*)data + 2*src;
+            int count = nseq * nchan;
+#if defined BF_AVX_ENABLED && BF_AVX_ENABLED
+            __m256i zero = _mm256_setzero_si256();
+            for( int i=0; i<count; ++i ) {
+                   _mm256_stream_si256(reinterpret_cast<__m256i*>(ptr+0), zero);
+                   _mm256_stream_si256(reinterpret_cast<__m256i*>(ptr+1), zero);
+                   ptr += 2*nsrc;
+            }
+#else
+            for( int i=0; i<count; ++i ) {
+                   ::memset(ptr, 0, 2*sizeof(otype));
+                   ptr += 2*nsrc;
             }
+#endif
     }
 };
 

src/formats/tbx.hpp

@@ -0,0 +1,63 @@
+/*
+ * Copyright (c) 2025, The Bifrost Authors. All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions
+ * are met:
+ * * Redistributions of source code must retain the above copyright
+ *   notice, this list of conditions and the following disclaimer.
+ * * Redistributions in binary form must reproduce the above copyright
+ *   notice, this list of conditions and the following disclaimer in the
+ *   documentation and/or other materials provided with the distribution.
+ * * Neither the name of The Bifrost Authors nor the names of its
+ *   contributors may be used to endorse or promote products derived
+ *   from this software without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS ``AS IS'' AND ANY
+ * EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+ * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
+ * PURPOSE ARE DISCLAIMED.  IN NO EVENT SHALL THE COPYRIGHT OWNER OR
+ * CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,
+ * EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
+ * PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
+ * PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY
+ * OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ */
+
+#pragma once
+
+#include "base.hpp"
+
+struct __attribute__((packed)) tbx_hdr_type {
+    uint32_t sync_word;
+    uint32_t frame_count_word;
+    uint32_t seconds_count;
+    uint32_t first_chan;
+    uint16_t nstand;
+    uint16_t nchan;
+    uint64_t time_tag;
+};
+
+template<int16_t NSTAND>
+class TBXHeaderFiller : virtual public PacketHeaderFiller {
+    int16_t            _nstand = NSTAND;
+public:
+    inline int get_size() { return sizeof(tbx_hdr_type); }
+    inline void operator()(const PacketDesc* hdr_base,
+                           BFoffset          framecount,
+                           char*             hdr) {
+        tbx_hdr_type* header = reinterpret_cast<tbx_hdr_type*>(hdr);
+        memset(header, 0, sizeof(tbx_hdr_type));
+
+        header->sync_word        = 0x5CDEC0DE;
+        // Bits 9-32 are the frame count; bits 1-8 are the TBX packet flag
+        header->frame_count_word = htobe32((framecount & 0xFFFFFF) \
+                                           | ((uint32_t) 0x08 << 24));
+        header->first_chan       = htobe32(hdr_base->src);
+        header->nstand           = htons(_nstand);
+        header->nchan            = htons(hdr_base->nchan);
+        header->time_tag         = htobe64(hdr_base->seq);
+    }
+};