Implement RX interrupt handling using gpio-cdev.

Sending still uses busy waiting, using interrupts there requires some
additional restructuring.

base-station/software/Cargo.lock

@@ -35,9 +35,11 @@ dependencies = [
  "env_logger",
  "futures",
  "futures-util",
+ "gpio-cdev",
  "linux-embedded-hal",
  "log",
  "mqtt-protocol",
+ "nb",
  "protocol",
  "rand",
  "sysfs_gpio",
@@ -109,10 +111,11 @@ dependencies = [
 [[package]]
 name = "embedded-nrf24l01"
 version = "0.2.0"
-source = "git+https://github.com/mgottschlag/embedded-nrf24l01?branch=wip#73fe8c9d16a3501c3e11b1320f66750643798d54"
+source = "git+https://github.com/mgottschlag/embedded-nrf24l01?branch=wip#e2d5148d169973550dade11456d6c9b2f58577eb"
 dependencies = [
  "bitfield",
  "embedded-hal",
+ "nb",
 ]
 
 [[package]]
@@ -256,6 +259,19 @@ dependencies = [
  "wasi",
 ]
 
+[[package]]
+name = "gpio-cdev"
+version = "0.3.0"
+source = "git+https://github.com/mgottschlag/gpio-cdev?branch=tokio#21d1704319fd747f67276149afbd7a285700de20"
+dependencies = [
+ "bitflags",
+ "futures",
+ "libc",
+ "mio",
+ "nix",
+ "tokio",
+]
+
 [[package]]
 name = "hermit-abi"
 version = "0.1.11"

base-station/software/Cargo.toml

@@ -20,6 +20,8 @@ tokio = { version = "0.2", features = [ "full" ] }
 sysfs_gpio = { version = "0.5.4", features = [ "use_tokio" ] }
 futures = "0.3.5"
 futures-util = "0.3.5"
+nb = "0.1.2"
+gpio-cdev = { git = "https://github.com/mgottschlag/gpio-cdev", branch = "tokio", features = [ "async-tokio" ] }
 
 [patch.crates-io]
 sysfs_gpio = { git = "https://github.com/mgottschlag/rust-sysfs-gpio", branch = "new-futures" }

base-station/software/src/radio.rs

@@ -8,8 +8,9 @@ use embedded_hal::blocking::spi::Transfer;
 use embedded_hal::digital::v2::OutputPin;
 use embedded_nrf24l01::{Configuration, CrcMode, DataRate, RxMode, NRF24L01};
 use futures_util::stream::StreamExt;
+use gpio_cdev::{AsyncLineEventHandle, Chip, EventRequestFlags, LineRequestFlags};
 use linux_embedded_hal::spidev::{SpiModeFlags, Spidev, SpidevOptions};
-use linux_embedded_hal::sysfs_gpio::{Direction, Edge, Error as GpioError};
+use linux_embedded_hal::sysfs_gpio::{Direction, Error as GpioError};
 use linux_embedded_hal::Pin;
 use log::{error, info};
 use protocol::{Location, Packet};
@@ -77,7 +78,7 @@ impl Radio {
 struct RadioTask {
     config: RadioConfig,
     nrf24: Option<RxMode<NRF24L01<GpioError, Pin, Pin, EmbeddedHalSpidev>>>,
-    irq: Option<Pin>,
+    irq: Option<AsyncLineEventHandle>,
     tx: UnboundedReceiver<(u8, Packet)>,
     rx: UnboundedSender<(u8, Packet)>,
 }
@@ -109,11 +110,14 @@ impl RadioTask {
         cs.export().unwrap();
         cs.set_direction(Direction::Out).unwrap();
         cs.set_high().unwrap();
+        let mut chip = Chip::new("/dev/gpiochip0")?;
         let irq_nr = get_pin_number('G', 9);
-        let irq = Pin::new(irq_nr);
-        irq.export().unwrap();
-        irq.set_direction(Direction::In).unwrap();
-        irq.set_edge(Edge::FallingEdge)?;
+        let line = chip.get_line(irq_nr as u32)?;
+        let irq = AsyncLineEventHandle::new(line.events(
+            LineRequestFlags::INPUT,
+            EventRequestFlags::BOTH_EDGES,
+            "gpioevents",
+        )?)?;
 
         // Configure SPI.
         let mut spi = Spidev::open("/dev/spidev0.0").unwrap();
@@ -139,7 +143,7 @@ impl RadioTask {
                     break;
                 }
             }
-            println!("radio not yet responding correctly ({:x})...", read_aw[1]);
+            info!("radio not yet responding correctly ({:x})...", read_aw[1]);
             delay_for(Duration::from_millis(1000)).await;
         }
 
@@ -183,10 +187,9 @@ impl RadioTask {
 
     async fn run_with_initialized_radio(&mut self) -> Result<(), RadioError> {
         // Wait for an RX interrupt of until we have to send a packet.
-        let mut irq_states = self.irq.as_mut().unwrap().get_value_stream().unwrap();
         loop {
             select! {
-                _pin_value = irq_states.next() => {
+                _pin_event = self.irq.as_mut().unwrap().next() => {
                     // Try to receive until there are no packets left in the RX fifo and until the
                     // interrupt pin is high again.
                     self.receive_packets().await?;
@@ -201,26 +204,26 @@ impl RadioTask {
     }
 
     async fn receive_packets(&mut self) -> Result<(), RadioError> {
-        let rx = self.nrf24.as_mut().unwrap();
-        let pipe = match rx.can_read()? {
-            Some(p) => p,
-            None => return Ok(()),
-        };
-        let payload = rx.read()?;
-        info!(
-            "packet received on pipe {}: {:x?}, {}",
-            pipe,
-            payload.as_ref(),
-            payload.len()
-        );
-        if payload.len() != 32 {
-            return Ok(());
-        }
-
-        let mut payload: [u8; 32] = payload.as_ref().try_into().unwrap();
-        self.handle_packet(&mut payload).await;
+        loop {
+            let rx = self.nrf24.as_mut().unwrap();
+            let pipe = match rx.can_read()? {
+                Some(p) => p,
+                None => return Ok(()),
+            };
+            let payload = rx.read()?;
+            info!(
+                "packet received on pipe {}: {:x?}, {}",
+                pipe,
+                payload.as_ref(),
+                payload.len()
+            );
+            if payload.len() != 32 {
+                return Ok(());
+            }
 
-        Ok(())
+            let mut payload: [u8; 32] = payload.as_ref().try_into().unwrap();
+            self.handle_packet(&mut payload).await;
+        }
     }
 
     async fn handle_packet(&mut self, packet: &mut [u8]) {
@@ -256,6 +259,7 @@ impl RadioTask {
 
     async fn send_packet(&mut self, device_id: u8, packet: Packet) -> Result<(), RadioError> {
         let salt = rand::thread_rng().gen::<u64>();
+        // TODO: Fix salt (device bit, device id).
         let mut tx = self
             .nrf24
             .take()
@@ -265,13 +269,31 @@ impl RadioTask {
             .map_err(|(_, e)| RadioError::Radio(e))?;
         let mut encoded = [0u8; 32];
         let key = get_key(device_id);
-        // TODO: Check whether the packet arrived.
+
+        info!("sending packet: {:?}", packet);
         if packet.encode_and_encrypt(key.unwrap(), salt, &mut encoded) {
             tx.set_tx_addr(&[0xB3, 0xB3, 0xB3, 0xB3, device_id])
                 .unwrap();
             tx.set_rx_addr(0, &[0xB3, 0xB3, 0xB3, 0xB3, device_id])
                 .unwrap();
             tx.send(&encoded).unwrap();
+
+            // Wait for the packet to be sent and check whether it was acknowledged.
+            loop {
+                delay_for(Duration::from_millis(1)).await;
+                match tx.poll_send() {
+                    Ok(status) => {
+                        if !status {
+                            error!("packet lost.");
+                        }
+                        break;
+                    }
+                    Err(nb::Error::WouldBlock) => {
+                        // TODO: Wait for interrupt.
+                    }
+                    Err(nb::Error::Other(e)) => return Err(e.into()),
+                }
+            }
         } else {
             info!("could not encode packet {:?}", packet);
         }
@@ -281,7 +303,6 @@ impl RadioTask {
                 .rx()
                 .map_err(|(_, e)| RadioError::Radio(e))?,
         );
-        info!("packet sent: {:?}", packet);
         Ok(())
     }
 }
@@ -292,6 +313,8 @@ pub enum RadioError {
     Io(#[from] io::Error),
     #[error("GPIO error")]
     Gpio(#[from] GpioError),
+    #[error("GPIO error (gpio-cdev)")]
+    GpioCdev(#[from] gpio_cdev::errors::Error),
     #[error("radio error: {0:?}")]
     Radio(embedded_nrf24l01::Error<std::io::Error>),
 }