Merge pull request #4 from sdr-enthusiasts/performance-improvements

Performance improvements

.gitignore

@@ -181,3 +181,5 @@ Cargo.lock
 
 #/target
 flamegraph.svg
+*.stacks
+cargo-flamegraph.stacks

Cargo.toml

@@ -11,7 +11,7 @@ members = [
 
 [workspace.package]
 edition = "2021"
-version = "0.1.0"
+version = "0.1.1"
 authors = ["Fred Clausen"]
 description = "ACARS Oxide. A utility to receive, via librtlsdr and RTL-SDR dongle(s), ACARS and VDLM2 messages."
 documentation = "https://github.com/sdr-enthusiasts/acars-oxide"
@@ -20,3 +20,9 @@ repository = "https://github.com/sdr-enthusiasts/acars-oxide"
 readme = "README.md"
 license = "MIT"
 rust-version = "1.69.0"
+
+# [profile.release]
+# debug = true
+# lto = "fat"
+# codegen-units = 1
+# opt-level = 2

rust/oxide-bin/Cargo.toml

@@ -19,3 +19,4 @@ oxide-logging = { path = "../oxide-logging" }
 oxide-scanner = { path = "../oxide-scanner" }
 oxide-decoders = { path = "../oxide-decoders" }
 tokio = { version = "1.28.1", features = ["full", "tracing"] }
+array-init = "2.1.0"

rust/oxide-bin/src/main.rs

@@ -246,7 +246,30 @@ async fn main() {
         }
     }
 
-    let scanner = oxide_scanner::OxideScanner::new(rtlsdr, args.output_to_console, false);
+    // FIXME: Fucked up padding of useless data
+    // Trying to avoid as much heap allocation....not sure I'm actually doing anything useful here
+    // pad the end of the vector with empty SDRs
+
+    let sdr_len: usize = rtlsdr.len();
+
+    while rtlsdr.len() < 8 {
+        rtlsdr.push(RtlSdr::new(
+            String::from(""),
+            0,
+            0,
+            false,
+            160,
+            vec![],
+            ValidDecoderType::ACARS,
+        ));
+    }
+
+    // create an array of length 8 to hold the SDRs
+
+    let rtlsdr_array: [RtlSdr; 8] = array_init::from_iter(rtlsdr.into_iter()).unwrap();
+
+    let scanner =
+        oxide_scanner::OxideScanner::new(rtlsdr_array, sdr_len, args.output_to_console, false);
     scanner.run().await;
 
     trace!("Starting the sleep loop");

rust/oxide-decoders/Cargo.toml

@@ -16,3 +16,4 @@ log = "0.4.17"
 custom_error = "1.9.2"
 num = "0.4.0"
 tokio = { version = "1.28.1", features = ["full", "tracing"] }
+# num-complex = "0.4.3"

rust/oxide-decoders/src/decoders/acars.rs

@@ -16,6 +16,7 @@
 
 use crate::Decoder;
 use custom_error::custom_error;
+// use num_complex::Complex;
 use num::Complex;
 use std::fmt::Display;
 use std::fmt::Formatter;
@@ -425,25 +426,25 @@ pub struct ACARSDecoder {
 }
 
 impl Decoder for ACARSDecoder {
-    fn decode(&mut self, length: u32) {
+    fn decode(&mut self, length: usize) {
         self.demod_msk(length);
     }
 
-    fn get_wf_at_index(&self, index: usize) -> Complex<f32> {
-        self.wf[index]
-    }
-
     fn set_dm_buffer_at_index(&mut self, index: usize, value: f32) {
         self.dm_buffer[index] = value;
     }
 
     fn set_output_channel(&mut self, output_channel: UnboundedSender<AssembledACARSMessage>) {
         self.output_channel = Some(output_channel);
     }
+
+    fn get_wf_iter(&self) -> std::slice::Iter<'_, Complex<f32>> {
+        self.wf.iter()
+    }
 }
 
 impl ACARSDecoder {
-    pub fn new(channel_number: i32, freq: i32, wf: [num::Complex<f32>; 192]) -> Self {
+    pub fn new(channel_number: i32, freq: i32, wf: [Complex<f32>; 192]) -> Self {
         let mut h: [f32; FLENO] = [0.0; FLENO];
         for (i, h_item) in h.iter_mut().enumerate().take(FLENO) {
             *h_item = f32::cos(
@@ -467,7 +468,7 @@ impl ACARSDecoder {
             msk_bit_count: 0,
             msk_s: 0,
             idx: 0,
-            inb: [num::Complex::new(0.0, 0.0); FLEN as usize],
+            inb: [Complex::new(0.0, 0.0); FLEN as usize],
             outbits: 0,
             nbits: 8,
             acars_state: ACARSState::Wsyn,
@@ -477,13 +478,12 @@ impl ACARSDecoder {
         }
     }
 
-    pub fn demod_msk(&mut self, len: u32) {
+    pub fn demod_msk(&mut self, len: usize) {
         /* MSK demod */
 
-        for n in 0..len as usize {
-            let in_: f32 = self.dm_buffer[n];
+        for in_ in &mut self.dm_buffer.into_iter().take(len) {
             let s: f32 = 1800.0 / INTRATE as f32 * 2.0 * std::f32::consts::PI + self.msk_df;
-            let mut v: num::Complex<f32> = num::Complex::new(0.0, 0.0);
+            let mut v: Complex<f32> = Complex::new(0.0, 0.0);
             let mut o: f32;
 
             /* VCO */
@@ -494,8 +494,7 @@ impl ACARSDecoder {
 
             /* mixer */
 
-            self.inb[self.idx as usize] =
-                in_ * num::Complex::exp(-self.msk_phi * num::Complex::i());
+            self.inb[self.idx as usize] = in_ * Complex::exp(-self.msk_phi * Complex::i());
             self.idx = (self.idx + 1) % (FLEN as u32);
 
             /* bit clock */

rust/oxide-decoders/src/lib.rs

@@ -15,6 +15,8 @@
 // Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA
 
 use decoders::acars::AssembledACARSMessage;
+//use num_complex::Complex;
+use num::Complex;
 use tokio::sync::mpsc::UnboundedSender;
 
 #[macro_use]
@@ -31,8 +33,8 @@ pub enum ValidDecoderType {
 }
 
 pub trait Decoder: Send + Sync {
-    fn decode(&mut self, length: u32);
-    fn get_wf_at_index(&self, index: usize) -> num::Complex<f32>;
+    fn decode(&mut self, length: usize);
+    fn get_wf_iter(&self) -> std::slice::Iter<'_, Complex<f32>>;
     fn set_dm_buffer_at_index(&mut self, index: usize, value: f32);
     fn set_output_channel(&mut self, channel: UnboundedSender<AssembledACARSMessage>);
 }

rust/oxide-rtlsdr/Cargo.toml

@@ -21,3 +21,5 @@ num = "0.4.0"
 custom_error = "1.9.2"
 oxide-decoders = { path = "../oxide-decoders" }
 tokio = { version = "1.28.1", features = ["full", "tracing"] }
+array-init = "2.1.0"
+# num-complex = "0.4.3"

rust/oxide-rtlsdr/src/lib.rs

@@ -16,6 +16,8 @@
 
 #[macro_use]
 extern crate log;
+// use num_complex::Complex;
+use num::Complex;
 use oxide_decoders::decoders::acars::ACARSDecoder;
 use oxide_decoders::decoders::acars::{self, AssembledACARSMessage};
 use oxide_decoders::{Decoder, ValidDecoderType};
@@ -47,7 +49,7 @@ pub struct RtlSdr {
     bias_tee: bool,
     rtl_mult: i32,
     frequencies: Vec<f32>,
-    channel: Vec<Box<dyn Decoder>>,
+    channel: [Box<dyn Decoder>; 16],
     decoder_type: ValidDecoderType,
 }
 
@@ -62,6 +64,17 @@ impl RtlSdr {
         decoder: ValidDecoderType,
     ) -> RtlSdr {
         frequencies.sort_by(|a, b| a.partial_cmp(b).unwrap());
+        let mut channels: Vec<Box<dyn Decoder>> = Vec::new();
+
+        // FIXME: This feels so wasteful to create 16 decoders when we only need 1 or 2
+        // But the array needs to be filled. Can we do better?
+        for _ in 0..16_usize {
+            channels.push(Box::new(ACARSDecoder::new(
+                0,
+                0,
+                [Complex::new(0.0, 0.0); 192],
+            )));
+        }
 
         Self {
             ctl: None,
@@ -73,7 +86,8 @@ impl RtlSdr {
             bias_tee,
             rtl_mult,
             frequencies,
-            channel: vec![],
+            // array_init::array_init(|i: usize| (i * i) as u32);
+            channel: array_init::from_iter(channels.into_iter()).unwrap(),
             decoder_type: decoder,
         }
     }
@@ -230,10 +244,10 @@ impl RtlSdr {
                     let am_freq =
                         ((channel - center_freq_actual) as f32) * 2.0 * std::f32::consts::PI
                             / (rtl_in_rate as f32);
-                    let mut window: Vec<num::Complex<f32>> = vec![];
+                    let mut window: Vec<Complex<f32>> = vec![];
                     for i in 0..self.rtl_mult {
                         // ch->wf[ind]=cexpf(AMFreq*ind*-I)/rtlMult/127.5;
-                        let window_value = (am_freq * i as f32 * -num::complex::Complex::i()).exp()
+                        let window_value = (am_freq * i as f32 * -Complex::i()).exp()
                             / self.rtl_mult as f32
                             / 127.5;
                         window.push(window_value);
@@ -243,16 +257,15 @@ impl RtlSdr {
 
                 for i in 0..self.frequencies.len() {
                     // create an array out of the channel_window[i] vector
-                    let mut window_array: [num::Complex<f32>; 192] =
-                        [num::complex::Complex::new(0.0, 0.0); 192];
+                    let mut window_array: [Complex<f32>; 192] = [Complex::new(0.0, 0.0); 192];
                     for (ind, window_value) in channel_windows[i].iter().enumerate() {
                         window_array[ind] = *window_value;
                     }
                     let mut out_channel: ACARSDecoder =
                         ACARSDecoder::new(i as i32, channels[i], window_array);
                     out_channel.set_output_channel(output_channel.clone());
 
-                    self.channel.push(Box::new(out_channel));
+                    self.channel[i] = Box::new(out_channel);
                 }
 
                 info!(
@@ -282,8 +295,7 @@ impl RtlSdr {
     pub async fn read_samples(mut self) {
         let rtloutbufz = self.get_rtloutbufsz();
         let buffer_len: u32 = rtloutbufz as u32 * self.rtl_mult as u32 * 2;
-        let mut vb: [num::Complex<f32>; 320] = [num::complex::Complex::new(0.0, 0.0); 320];
-        let mut counter: usize = 0;
+        let mut vb: [Complex<f32>; 320] = [Complex::new(0.0, 0.0); 320];
 
         match self.reader {
             None => {
@@ -293,29 +305,38 @@ impl RtlSdr {
             Some(mut reader) => {
                 reader
                     .read_async(4, buffer_len, |bytes: &[u8]| {
-                        counter = 0;
+                        let mut bytes_iterator = bytes.iter();
+
                         for m in 0..rtloutbufz {
                             for vb_item in vb.iter_mut().take(self.rtl_mult as usize) {
-                                *vb_item = (bytes[counter] as f32 - 127.37)
-                                    + (bytes[counter + 1] as f32 - 127.37)
-                                        * num::complex::Complex::i();
-                                counter += 2;
+                                *vb_item = (bytes_iterator
+                                    .next()
+                                    .expect("Ran out of bytes!")
+                                    .to_owned() as f32
+                                    - 127.37)
+                                    + (bytes_iterator.next().expect("Ran out of bytes!").to_owned()
+                                        as f32
+                                        - 127.37)
+                                        * Complex::i();
                             }
 
-                            for channel in &mut self.channel {
-                                let mut d: num::Complex<f32> = num::complex::Complex::new(0.0, 0.0);
+                            for channel in &mut self.channel.iter_mut().take(self.frequencies.len())
+                            {
+                                let mut d: Complex<f32> = Complex::new(0.0, 0.0);
 
-                                for (ind, vb_item) in
-                                    vb.iter().enumerate().take(self.rtl_mult as usize)
+                                for (wf, vb_item) in vb
+                                    .iter()
+                                    .zip(channel.get_wf_iter())
+                                    .take(self.rtl_mult as usize)
                                 {
-                                    d += vb_item * channel.get_wf_at_index(ind);
+                                    d += vb_item * wf;
                                 }
 
                                 channel.set_dm_buffer_at_index(m, d.norm());
                             }
                         }
-                        for channel in &mut self.channel {
-                            channel.decode(rtloutbufz as u32);
+                        for channel in &mut self.channel.iter_mut().take(self.frequencies.len()) {
+                            channel.decode(rtloutbufz);
                         }
                     })
                     .unwrap();

rust/oxide-scanner/src/lib.rs

@@ -22,19 +22,22 @@ use oxide_rtlsdr::RtlSdr;
 use tokio::sync::mpsc;
 
 pub struct OxideScanner {
-    sdrs: Vec<RtlSdr>,
+    sdrs: [RtlSdr; 8],
     enable_output_command_line: bool,
     enable_output_zmq: bool,
+    number_of_sdrs: usize,
 }
 
 impl OxideScanner {
     pub fn new(
-        sdrs: Vec<RtlSdr>,
+        sdrs: [RtlSdr; 8],
+        number_of_sdrs: usize,
         enable_output_command_line: bool,
         enable_output_zmq: bool,
     ) -> OxideScanner {
         OxideScanner {
             sdrs,
+            number_of_sdrs,
             enable_output_command_line,
             enable_output_zmq,
         }
@@ -50,7 +53,7 @@ impl OxideScanner {
             output.monitor_receiver_channel().await;
         });
 
-        for mut sdr in self.sdrs.into_iter() {
+        for mut sdr in self.sdrs.into_iter().take(self.number_of_sdrs) {
             info!("[OXIDE SCANNER] Opening SDR {}", sdr.get_serial());
             match sdr.open_sdr(tx_channel.clone()) {
                 Ok(_) => {