Multiple improvements to the exchange and transport layers (#147)

* Multiple improvements to the exchange and transport layers

Bugfixing in subscriptions

Mdns shares buffers with the main transport

Complete subscription logic (incl change notification)

Bugfixing

Bugfixing

Best effort for Google controller subscriptions to stay alive

Re-publish the mDNS broadcast when an entry is removed too

Google controller expects revoke commissioning to be supported

Eagerly close subscriptions that don't report anything

Cleanup in transport mgr

Restore the correct subscription id

Simplify transport mgr

Tests typecheck

Minor renames

Docu

Docu

Docu, clippy

fix the tests

Handle session close

Fix buffer sizes for subscriptions

Report responer memory

RFC (work in progress)

RFC (work in progress)

RFC (work in progress)

RFC (work in progress)

RFC (work in progress)

RFC (work in progress)

RFC (work in progress)

RFC (work in progress)

RFC (work in progress)

RFC (work in progress)

RFC (work in progress)

RFC (work in progress)

RFC (work in progress)

RFC (work in progress)

RFC (WIP)

Buffers for IM; more flexible Exchange; renames in Exchange

Address several bugs

Matter header for Notification

Address several bugs

Unit tests work

Address several bugs

Small updates to the RFC

Small updates to the RFC

Make IM compatible with unit tests

Enable operation over reliable protocols

Support for large buffers (TCP)

Unify the synchronization primitives

Try to reduce a bit the consumed memory

Reduce the change delta

Update RFC

Extra comments

Fix the build

WIP - RFC

Address several bugs

Fix lifetime issues with subscriptions notifications

Updates to the RFC

* Document the handler API

* Clarify a commented out line

* Document the await optimization

* Remove a level of indentation

* Leave a TODO that trhe subscription notification logic is incomplete

* Add a warning for an unanticipated opcode

* Skip the doctest

* Change semantics the of recv and recv_fetch to return the last fetched message, if there is any

* Address feedback from code review

* std::net not necessary as it is now just re-exporting core::net

* Address feedback from code review

* Address feedback from code review

* Address feedback from code review

* Address feedback from code review

* Address feedback from code review

* Address feedback from code review

* Address feedback from code review

* Address feedback from code review

* Address feedback from code review

* Address feedback from code review

* Incorporate changes to the RFC from HackMD

examples/onoff_light/src/dev_att.rs

@@ -21,7 +21,7 @@ use rs_matter::error::{Error, ErrorCode};
 pub struct HardCodedDevAtt {}
 
 impl HardCodedDevAtt {
-    pub fn new() -> Self {
+    pub const fn new() -> Self {
         Self {}
     }
 }

examples/onoff_light/src/main.rs

@@ -19,23 +19,29 @@ use core::borrow::Borrow;
 use core::pin::pin;
 use std::net::UdpSocket;
 
-use embassy_futures::select::select3;
+use embassy_futures::select::{select, select4};
 
+use embassy_sync::blocking_mutex::raw::NoopRawMutex;
+use embassy_time::{Duration, Timer};
 use log::info;
 
 use rs_matter::core::{CommissioningData, Matter};
 use rs_matter::data_model::cluster_basic_information::BasicInfoConfig;
 use rs_matter::data_model::cluster_on_off;
+use rs_matter::data_model::core::IMBuffer;
 use rs_matter::data_model::device_types::DEV_TYPE_ON_OFF_LIGHT;
 use rs_matter::data_model::objects::*;
 use rs_matter::data_model::root_endpoint;
+use rs_matter::data_model::subscriptions::Subscriptions;
 use rs_matter::data_model::system_model::descriptor;
 use rs_matter::error::Error;
 use rs_matter::mdns::MdnsService;
 use rs_matter::persist::Psm;
+use rs_matter::respond::DefaultResponder;
 use rs_matter::secure_channel::spake2p::VerifierData;
-use rs_matter::transport::core::{PacketBuffers, MATTER_SOCKET_BIND_ADDR};
-use rs_matter::utils::select::EitherUnwrap;
+use rs_matter::transport::core::MATTER_SOCKET_BIND_ADDR;
+use rs_matter::utils::buf::PooledBuffers;
+use rs_matter::utils::select::Coalesce;
 use rs_matter::MATTER_PORT;
 
 mod dev_att;
@@ -49,7 +55,7 @@ fn main() -> Result<(), Error> {
         // e.g., an opt-level of "0" will require a several times' larger stack.
         //
         // Optimizing/lowering `rs-matter` memory consumption is an ongoing topic.
-        .stack_size(180 * 1024)
+        .stack_size(95 * 1024)
         .spawn(run)
         .unwrap();
 
@@ -62,9 +68,9 @@ fn run() -> Result<(), Error> {
     );
 
     info!(
-        "Matter memory: Matter={}, PacketBuffers={}",
+        "Matter memory: Matter={}B, IM Buffers={}B",
         core::mem::size_of::<Matter>(),
-        core::mem::size_of::<PacketBuffers>(),
+        core::mem::size_of::<PooledBuffers<10, NoopRawMutex, IMBuffer>>()
     );
 
     let dev_det = BasicInfoConfig {
@@ -81,57 +87,92 @@ fn run() -> Result<(), Error> {
 
     let dev_att = dev_att::HardCodedDevAtt::new();
 
-    // NOTE:
-    // For `no_std` environments, provide your own epoch and rand functions here
-    let epoch = rs_matter::utils::epoch::sys_epoch;
-    let rand = rs_matter::utils::rand::sys_rand;
-
     let matter = Matter::new(
-        // vid/pid should match those in the DAC
         &dev_det,
         &dev_att,
+        // NOTE:
+        // For `no_std` environments, provide your own epoch and rand functions here
         MdnsService::Builtin,
-        epoch,
-        rand,
+        rs_matter::utils::epoch::sys_epoch,
+        rs_matter::utils::rand::sys_rand,
         MATTER_PORT,
     );
 
+    matter.initialize_transport_buffers()?;
+
     info!("Matter initialized");
 
-    let handler = HandlerCompat(handler(&matter));
+    let buffers = PooledBuffers::<10, NoopRawMutex, _>::new(0);
+
+    info!("IM buffers initialized");
+
+    let mut mdns = pin!(run_mdns(&matter));
+
+    let on_off = cluster_on_off::OnOffCluster::new(*matter.borrow());
+
+    let subscriptions = Subscriptions::<3>::new();
+
+    // Assemble our Data Model handler by composing the predefined Root Endpoint handler with our custom On/Off clusters
+    let dm_handler = HandlerCompat(dm_handler(&matter, &on_off));
+
+    // Create a default responder capable of handling up to 3 subscriptions
+    // All other subscription requests will be turned down with "resource exhausted"
+    let responder = DefaultResponder::new(&matter, &buffers, &subscriptions, dm_handler);
+    info!(
+        "Responder memory: Responder={}B, Runner={}B",
+        core::mem::size_of_val(&responder),
+        core::mem::size_of_val(&responder.run::<4, 4>())
+    );
+
+    // Run the responder with up to 4 handlers (i.e. 4 exchanges can be handled simultenously)
+    // Clients trying to open more exchanges than the ones currently running will get "I'm busy, please try again later"
+    let mut respond = pin!(responder.run::<4, 4>());
+
+    // This is a sample code that simulates state changes triggered by the HAL
+    // Changes will be properly communicated to the Matter controllers and other Matter apps (i.e. Google Home, Alexa), thanks to subscriptions
+    let mut device = pin!(async {
+        loop {
+            Timer::after(Duration::from_secs(5)).await;
+
+            on_off.set(!on_off.get());
+            subscriptions.notify_changed();
+
+            info!("Lamp toggled");
+        }
+    });
 
     // NOTE:
     // When using a custom UDP stack (e.g. for `no_std` environments), replace with a UDP socket bind for your custom UDP stack
     // The returned socket should be splittable into two halves, where each half implements `UdpSend` and `UdpReceive` respectively
     let socket = async_io::Async::<UdpSocket>::bind(MATTER_SOCKET_BIND_ADDR)?;
 
-    let mut packet_buffers = PacketBuffers::new();
-    let mut runner = pin!(matter.run(
+    // Run the Matter and mDNS transports
+    let mut transport = pin!(matter.run(
         &socket,
         &socket,
-        &mut packet_buffers,
-        CommissioningData {
+        Some(CommissioningData {
             // TODO: Hard-coded for now
             verifier: VerifierData::new_with_pw(123456, *matter.borrow()),
             discriminator: 250,
-        },
-        &handler,
+        }),
     ));
 
-    let mut mdns_runner = pin!(run_mdns(&matter));
-
     // NOTE:
     // Replace with your own persister for e.g. `no_std` environments
     let mut psm = Psm::new(&matter, std::env::temp_dir().join("rs-matter"))?;
-    let mut psm_runner = pin!(psm.run());
-
-    let runner = select3(&mut runner, &mut mdns_runner, &mut psm_runner);
+    let mut persist = pin!(psm.run());
+
+    // Combine all async tasks in a single one
+    let all = select4(
+        &mut transport,
+        &mut mdns,
+        &mut persist,
+        select(&mut respond, &mut device).coalesce(),
+    );
 
     // NOTE:
     // Replace with a different executor for e.g. `no_std` environments
-    futures_lite::future::block_on(runner).unwrap()?;
-
-    Ok(())
+    futures_lite::future::block_on(all.coalesce())
 }
 
 const NODE: Node<'static> = Node {
@@ -146,7 +187,10 @@ const NODE: Node<'static> = Node {
     ],
 };
 
-fn handler<'a>(matter: &'a Matter<'a>) -> impl Metadata + NonBlockingHandler + 'a {
+fn dm_handler<'a>(
+    matter: &'a Matter<'a>,
+    on_off: &'a cluster_on_off::OnOffCluster,
+) -> impl Metadata + NonBlockingHandler + 'a {
     (
         NODE,
         root_endpoint::handler(0, matter)
@@ -155,11 +199,7 @@ fn handler<'a>(matter: &'a Matter<'a>) -> impl Metadata + NonBlockingHandler + '
                 descriptor::ID,
                 descriptor::DescriptorCluster::new(*matter.borrow()),
             )
-            .chain(
-                1,
-                cluster_on_off::ID,
-                cluster_on_off::OnOffCluster::new(*matter.borrow()),
-            ),
+            .chain(1, cluster_on_off::ID, on_off),
     )
 }
 

rs-matter/Cargo.toml

@@ -20,6 +20,7 @@ alloc = []
 openssl = ["alloc", "dep:openssl", "foreign-types", "hmac", "sha2"]
 mbedtls = ["alloc", "dep:mbedtls"]
 rustcrypto = ["alloc", "sha2", "hmac", "pbkdf2", "hkdf", "aes", "ccm", "p256", "elliptic-curve", "crypto-bigint", "x509-cert", "rand_core"]
+large-buffers = [] # TCP support
 
 [dependencies]
 rs-matter-macros = { version = "0.1", path = "../rs-matter-macros" }
@@ -45,6 +46,7 @@ domain = { version = "0.9", default-features = false, features = ["heapless"] }
 octseq = { version = "0.3", default-features = false }
 portable-atomic = "1"
 qrcodegen-no-heap = "1.8"
+scopeguard = "1"
 
 # crypto
 openssl = { version = "0.10", optional = true }