LLmp hooks (#2280)

* llmp hooks

* separate llmp broker from inner state to allow doing more stuff

libafl/src/events/centralized.rs

@@ -8,7 +8,9 @@
 // 4. The "main broker", the gathers the stats from the fuzzer clients and broadcast the newly found testcases from the main evaluator.
 
 use alloc::{boxed::Box, string::String, vec::Vec};
-use core::{marker::PhantomData, num::NonZeroUsize, time::Duration};
+use core::fmt::Debug;
+#[cfg(feature = "adaptive_serialization")]
+use core::time::Duration;
 
 #[cfg(feature = "adaptive_serialization")]
 use libafl_bolts::tuples::{Handle, Handled};
@@ -18,7 +20,7 @@ use libafl_bolts::{
     llmp::{LLMP_FLAG_COMPRESSED, LLMP_FLAG_INITIALIZED},
 };
 use libafl_bolts::{
-    llmp::{self, LlmpBroker, LlmpClient, LlmpClientDescription, Tag},
+    llmp::{LlmpClient, LlmpClientDescription, Tag},
     shmem::{NopShMemProvider, ShMemProvider},
     ClientId,
 };
@@ -33,9 +35,9 @@ use crate::observers::TimeObserver;
 use crate::state::HasScalabilityMonitor;
 use crate::{
     events::{
-        AdaptiveSerializer, BrokerEventResult, CustomBufEventResult, Event, EventConfig,
+        AdaptiveSerializer, CustomBufEventResult, Event, EventConfig, EventFirer, EventManager,
-        EventFirer, EventManager, EventManagerId, EventProcessor, EventRestarter,
+        EventManagerId, EventProcessor, EventRestarter, HasCustomBufHandlers, HasEventManagerId,
-        HasCustomBufHandlers, HasEventManagerId, LogSeverity, ProgressReporter,
+        LogSeverity, ProgressReporter,
     },
     executors::{Executor, HasObservers},
     fuzzer::{EvaluatorObservers, ExecutionProcessor},
@@ -45,174 +47,7 @@ use crate::{
     Error, HasMetadata,
 };
 
-const _LLMP_TAG_TO_MAIN: Tag = Tag(0x3453453);
+pub(crate) const _LLMP_TAG_TO_MAIN: Tag = Tag(0x3453453);
-
-/// An LLMP-backed event manager for scalable multi-processed fuzzing
-pub struct CentralizedLlmpEventBroker<I, SP>
-where
-    I: Input,
-    SP: ShMemProvider + 'static,
-    //CE: CustomEvent<I>,
-{
-    llmp: LlmpBroker<SP>,
-    #[cfg(feature = "llmp_compression")]
-    compressor: GzipCompressor,
-    phantom: PhantomData<I>,
-}
-
-impl<I, SP> core::fmt::Debug for CentralizedLlmpEventBroker<I, SP>
-where
-    SP: ShMemProvider + 'static,
-    I: Input,
-{
-    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
-        let mut debug_struct = f.debug_struct("CentralizedLlmpEventBroker");
-        let debug = debug_struct.field("llmp", &self.llmp);
-        //.field("custom_buf_handlers", &self.custom_buf_handlers)
-        #[cfg(feature = "llmp_compression")]
-        let debug = debug.field("compressor", &self.compressor);
-        debug
-            .field("phantom", &self.phantom)
-            .finish_non_exhaustive()
-    }
-}
-
-impl<I, SP> CentralizedLlmpEventBroker<I, SP>
-where
-    I: Input,
-    SP: ShMemProvider + 'static,
-{
-    /// Create an event broker from a raw broker.
-    pub fn new(llmp: LlmpBroker<SP>) -> Result<Self, Error> {
-        Ok(Self {
-            llmp,
-            #[cfg(feature = "llmp_compression")]
-            compressor: GzipCompressor::with_threshold(COMPRESS_THRESHOLD),
-            phantom: PhantomData,
-        })
-    }
-
-    /// Create an LLMP broker on a port.
-    ///
-    /// The port must not be bound yet to have a broker.
-    #[cfg(feature = "std")]
-    pub fn on_port(shmem_provider: SP, port: u16) -> Result<Self, Error> {
-        Ok(Self {
-            // TODO switch to false after solving the bug
-            llmp: LlmpBroker::with_keep_pages_attach_to_tcp(shmem_provider, port, true)?,
-            #[cfg(feature = "llmp_compression")]
-            compressor: GzipCompressor::with_threshold(COMPRESS_THRESHOLD),
-            phantom: PhantomData,
-        })
-    }
-
-    /// Exit the broker process cleanly after at least `n` clients attached and all of them disconnected again
-    pub fn set_exit_cleanly_after(&mut self, n_clients: NonZeroUsize) {
-        self.llmp.set_exit_cleanly_after(n_clients);
-    }
-
-    /// Run forever in the broker
-    #[cfg(not(feature = "llmp_broker_timeouts"))]
-    pub fn broker_loop(&mut self) -> Result<(), Error> {
-        #[cfg(feature = "llmp_compression")]
-        let compressor = &self.compressor;
-        self.llmp.loop_forever(
-            &mut |client_id, tag, _flags, msg| {
-                if tag == _LLMP_TAG_TO_MAIN {
-                    #[cfg(not(feature = "llmp_compression"))]
-                    let event_bytes = msg;
-                    #[cfg(feature = "llmp_compression")]
-                    let compressed;
-                    #[cfg(feature = "llmp_compression")]
-                    let event_bytes = if _flags & LLMP_FLAG_COMPRESSED == LLMP_FLAG_COMPRESSED {
-                        compressed = compressor.decompress(msg)?;
-                        &compressed
-                    } else {
-                        msg
-                    };
-                    let event: Event<I> = postcard::from_bytes(event_bytes)?;
-                    match Self::handle_in_broker(client_id, &event)? {
-                        BrokerEventResult::Forward => Ok(llmp::LlmpMsgHookResult::ForwardToClients),
-                        BrokerEventResult::Handled => Ok(llmp::LlmpMsgHookResult::Handled),
-                    }
-                } else {
-                    Ok(llmp::LlmpMsgHookResult::ForwardToClients)
-                }
-            },
-            Some(Duration::from_millis(5)),
-        );
-
-        #[cfg(all(feature = "std", feature = "llmp_debug"))]
-        println!("The last client quit. Exiting.");
-
-        Err(Error::shutting_down())
-    }
-
-    /// Run in the broker until all clients exit
-    #[cfg(feature = "llmp_broker_timeouts")]
-    pub fn broker_loop(&mut self) -> Result<(), Error> {
-        #[cfg(feature = "llmp_compression")]
-        let compressor = &self.compressor;
-        self.llmp.loop_with_timeouts(
-            &mut |msg_or_timeout| {
-                if let Some((client_id, tag, _flags, msg)) = msg_or_timeout {
-                    if tag == _LLMP_TAG_TO_MAIN {
-                        #[cfg(not(feature = "llmp_compression"))]
-                        let event_bytes = msg;
-                        #[cfg(feature = "llmp_compression")]
-                        let compressed;
-                        #[cfg(feature = "llmp_compression")]
-                        let event_bytes = if _flags & LLMP_FLAG_COMPRESSED == LLMP_FLAG_COMPRESSED {
-                            compressed = compressor.decompress(msg)?;
-                            &compressed
-                        } else {
-                            msg
-                        };
-                        let event: Event<I> = postcard::from_bytes(event_bytes)?;
-                        match Self::handle_in_broker(client_id, &event)? {
-                            BrokerEventResult::Forward => {
-                                Ok(llmp::LlmpMsgHookResult::ForwardToClients)
-                            }
-                            BrokerEventResult::Handled => Ok(llmp::LlmpMsgHookResult::Handled),
-                        }
-                    } else {
-                        Ok(llmp::LlmpMsgHookResult::ForwardToClients)
-                    }
-                } else {
-                    Ok(llmp::LlmpMsgHookResult::Handled)
-                }
-            },
-            Duration::from_secs(30),
-            Some(Duration::from_millis(5)),
-        );
-
-        #[cfg(feature = "llmp_debug")]
-        println!("The last client quit. Exiting.");
-
-        Err(Error::shutting_down())
-    }
-
-    /// Handle arriving events in the broker
-    #[allow(clippy::unnecessary_wraps)]
-    fn handle_in_broker(
-        _client_id: ClientId,
-        event: &Event<I>,
-    ) -> Result<BrokerEventResult, Error> {
-        match &event {
-            Event::NewTestcase {
-                input: _,
-                client_config: _,
-                exit_kind: _,
-                corpus_size: _,
-                observers_buf: _,
-                time: _,
-                executions: _,
-                forward_id: _,
-            } => Ok(BrokerEventResult::Forward),
-            _ => Ok(BrokerEventResult::Handled),
-        }
-    }
-}
 
 /// A wrapper manager to implement a main-secondary architecture with another broker
 #[derive(Debug)]

libafl/src/events/launcher.rs

@@ -28,6 +28,8 @@ use std::process::Stdio;
 #[cfg(all(unix, feature = "std"))]
 use std::{fs::File, os::unix::io::AsRawFd};
 
+#[cfg(all(unix, feature = "std", feature = "fork"))]
+use libafl_bolts::llmp::LlmpBroker;
 #[cfg(all(unix, feature = "std"))]
 use libafl_bolts::os::dup2;
 #[cfg(all(feature = "std", any(windows, not(feature = "fork"))))]
@@ -52,7 +54,7 @@ use super::hooks::EventManagerHooksTuple;
 use crate::observers::TimeObserver;
 #[cfg(all(unix, feature = "std", feature = "fork"))]
 use crate::{
-    events::centralized::{CentralizedEventManager, CentralizedLlmpEventBroker},
+    events::{centralized::CentralizedEventManager, llmp::centralized::CentralizedLlmpHook},
     state::UsesState,
 };
 #[cfg(feature = "std")]
@@ -167,8 +169,8 @@ where
 impl<'a, CF, MT, S, SP> Launcher<'a, CF, (), MT, S, SP>
 where
     CF: FnOnce(Option<S>, LlmpRestartingEventManager<(), S, SP>, CoreId) -> Result<(), Error>,
-    MT: Monitor + Clone,
+    MT: Monitor + Clone + 'static,
-    S: State + HasExecutions,
+    S: State + HasExecutions + 'static,
     SP: ShMemProvider + 'static,
 {
     /// Launch the broker and the clients and fuzz
@@ -190,8 +192,8 @@ impl<'a, CF, EMH, MT, S, SP> Launcher<'a, CF, EMH, MT, S, SP>
 where
     CF: FnOnce(Option<S>, LlmpRestartingEventManager<EMH, S, SP>, CoreId) -> Result<(), Error>,
     EMH: EventManagerHooksTuple<S> + Clone + Copy,
-    MT: Monitor + Clone,
+    MT: Monitor + Clone + 'static,
-    S: State + HasExecutions,
+    S: State + HasExecutions + 'static,
     SP: ShMemProvider + 'static,
 {
     /// Launch the broker and the clients and fuzz with a user-supplied hook
@@ -562,8 +564,8 @@ where
         CentralizedEventManager<StdCentralizedInnerMgr<S, SP>, SP>,
         CoreId,
     ) -> Result<(), Error>,
-    MT: Monitor + Clone,
+    MT: Monitor + Clone + 'static,
-    S: State + HasExecutions,
+    S: State + HasExecutions + 'static,
     SP: ShMemProvider + 'static,
 {
     /// Launch a standard Centralized-based fuzzer
@@ -601,8 +603,8 @@ where
         CentralizedEventManager<IM, SP>, // No hooks for centralized EM
         CoreId,
     ) -> Result<(), Error>,
-    MT: Monitor + Clone,
+    MT: Monitor + Clone + 'static,
-    S: State + HasExecutions,
+    S: State + HasExecutions + 'static,
     SP: ShMemProvider + 'static,
 {
     /// Launch a Centralized-based fuzzer.
@@ -663,12 +665,22 @@ where
                 log::info!("PID: {:#?} I am centralized broker", std::process::id());
                 self.shmem_provider.post_fork(true)?;
 
-                let mut broker: CentralizedLlmpEventBroker<S::Input, SP> =
+                let llmp_centralized_hook = CentralizedLlmpHook::<S::Input, SP>::new()?;
-                    CentralizedLlmpEventBroker::on_port(
+
-                        self.shmem_provider.clone(),
+                // TODO switch to false after solving the bug
-                        self.centralized_broker_port,
+                let mut broker = LlmpBroker::with_keep_pages_attach_to_tcp(
-                    )?;
+                    self.shmem_provider.clone(),
-                broker.broker_loop()?;
+                    tuple_list!(llmp_centralized_hook),
+                    self.centralized_broker_port,
+                    true,
+                )?;
+
+                // Run in the broker until all clients exit
+                broker.loop_with_timeouts(Duration::from_secs(30), Some(Duration::from_millis(5)));
+
+                log::info!("The last client quit. Exiting.");
+
+                return Err(Error::shutting_down());
             }
         }
 

libafl/src/events/llmp/hooks/centralized.rs

@@ -0,0 +1,110 @@
+use std::{fmt::Debug, marker::PhantomData};
+
+#[cfg(feature = "llmp_compression")]
+use libafl_bolts::{compress::GzipCompressor, llmp::LLMP_FLAG_COMPRESSED};
+use libafl_bolts::{
+    llmp::{Flags, LlmpBrokerState, LlmpHook, LlmpMsgHookResult, Tag},
+    shmem::ShMemProvider,
+    ClientId, Error,
+};
+
+#[cfg(feature = "llmp_compression")]
+use crate::events::COMPRESS_THRESHOLD;
+use crate::{
+    events::{BrokerEventResult, Event, _LLMP_TAG_TO_MAIN},
+    inputs::Input,
+};
+
+/// An LLMP-backed event manager for scalable multi-processed fuzzing
+pub struct CentralizedLlmpHook<I, SP> {
+    #[cfg(feature = "llmp_compression")]
+    compressor: GzipCompressor,
+    phantom: PhantomData<(I, SP)>,
+}
+
+impl<I, SP> LlmpHook<SP> for CentralizedLlmpHook<I, SP>
+where
+    I: Input,
+    SP: ShMemProvider + 'static,
+{
+    fn on_new_message(
+        &mut self,
+        _llmp_broker_state: &mut LlmpBrokerState<SP>,
+        client_id: ClientId,
+        msg_tag: &mut Tag,
+        msg_flags: &mut Flags,
+        msg: &mut [u8],
+    ) -> Result<LlmpMsgHookResult, Error> {
+        if *msg_tag == _LLMP_TAG_TO_MAIN {
+            #[cfg(feature = "llmp_compression")]
+            let compressor = &self.compressor;
+            #[cfg(not(feature = "llmp_compression"))]
+            let event_bytes = msg;
+            #[cfg(feature = "llmp_compression")]
+            let compressed;
+            #[cfg(feature = "llmp_compression")]
+            let event_bytes = if *msg_flags & LLMP_FLAG_COMPRESSED == LLMP_FLAG_COMPRESSED {
+                compressed = compressor.decompress(msg)?;
+                &compressed
+            } else {
+                &*msg
+            };
+            let event: Event<I> = postcard::from_bytes(event_bytes)?;
+            match Self::handle_in_broker(client_id, &event)? {
+                BrokerEventResult::Forward => Ok(LlmpMsgHookResult::ForwardToClients),
+                BrokerEventResult::Handled => Ok(LlmpMsgHookResult::Handled),
+            }
+        } else {
+            Ok(LlmpMsgHookResult::ForwardToClients)
+        }
+    }
+}
+
+impl<I, SP> Debug for CentralizedLlmpHook<I, SP> {
+    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
+        let mut debug_struct = f.debug_struct("CentralizedLlmpEventBroker");
+
+        #[cfg(feature = "llmp_compression")]
+        let debug_struct = debug_struct.field("compressor", &self.compressor);
+
+        debug_struct
+            .field("phantom", &self.phantom)
+            .finish_non_exhaustive()
+    }
+}
+
+impl<I, SP> CentralizedLlmpHook<I, SP>
+where
+    I: Input,
+    SP: ShMemProvider + 'static,
+{
+    /// Create an event broker from a raw broker.
+    pub fn new() -> Result<Self, Error> {
+        Ok(Self {
+            #[cfg(feature = "llmp_compression")]
+            compressor: GzipCompressor::with_threshold(COMPRESS_THRESHOLD),
+            phantom: PhantomData,
+        })
+    }
+
+    /// Handle arriving events in the broker
+    #[allow(clippy::unnecessary_wraps)]
+    fn handle_in_broker(
+        _client_id: ClientId,
+        event: &Event<I>,
+    ) -> Result<BrokerEventResult, Error> {
+        match &event {
+            Event::NewTestcase {
+                input: _,
+                client_config: _,
+                exit_kind: _,
+                corpus_size: _,
+                observers_buf: _,
+                time: _,
+                executions: _,
+                forward_id: _,
+            } => Ok(BrokerEventResult::Forward),
+            _ => Ok(BrokerEventResult::Handled),
+        }
+    }
+}

libafl/src/events/llmp/hooks/mod.rs

@@ -1,12 +1,13 @@
-//! LLMP broker
+//! Standard LLMP hook
-
+use core::marker::PhantomData;
-use core::{marker::PhantomData, num::NonZeroUsize, time::Duration};
-#[cfg(feature = "std")]
-use std::net::ToSocketAddrs;
 
 #[cfg(feature = "llmp_compression")]
 use libafl_bolts::{compress::GzipCompressor, llmp::LLMP_FLAG_COMPRESSED};
-use libafl_bolts::{llmp, shmem::ShMemProvider, ClientId};
+use libafl_bolts::{
+    llmp::{Flags, LlmpBrokerState, LlmpHook, LlmpMsgHookResult, Tag},
+    shmem::ShMemProvider,
+    ClientId,
+};
 
 #[cfg(feature = "llmp_compression")]
 use crate::events::llmp::COMPRESS_THRESHOLD;
@@ -17,148 +18,85 @@ use crate::{
     Error,
 };
 
-/// An LLMP-backed event manager for scalable multi-processed fuzzing
+/// centralized hook
+#[cfg(all(unix, feature = "std"))]
+pub mod centralized;
+
+/// An LLMP-backed event hook for scalable multi-processed fuzzing
 #[derive(Debug)]
-pub struct LlmpEventBroker<I, MT, SP>
+pub struct StdLlmpEventHook<I, MT, SP>
 where
     SP: ShMemProvider + 'static,
 {
     monitor: MT,
-    llmp: llmp::LlmpBroker<SP>,
     #[cfg(feature = "llmp_compression")]
     compressor: GzipCompressor,
-    phantom: PhantomData<I>,
+    phantom: PhantomData<(I, SP)>,
 }
 
-impl<I, MT, SP> LlmpEventBroker<I, MT, SP>
+impl<I, MT, SP> LlmpHook<SP> for StdLlmpEventHook<I, MT, SP>
+where
+    I: Input,
+    MT: Monitor,
+    SP: ShMemProvider + 'static,
+{
+    fn on_new_message(
+        &mut self,
+        _llmp_broker_state: &mut LlmpBrokerState<SP>,
+        client_id: ClientId,
+        msg_tag: &mut Tag,
+        #[cfg(feature = "llmp_compression")] msg_flags: &mut Flags,
+        #[cfg(not(feature = "llmp_compression"))] _msg_flags: &mut Flags,
+        msg: &mut [u8],
+    ) -> Result<LlmpMsgHookResult, Error> {
+        let monitor = &mut self.monitor;
+        #[cfg(feature = "llmp_compression")]
+        let compressor = &self.compressor;
+
+        if *msg_tag == LLMP_TAG_EVENT_TO_BOTH {
+            #[cfg(not(feature = "llmp_compression"))]
+            let event_bytes = msg;
+            #[cfg(feature = "llmp_compression")]
+            let compressed;
+            #[cfg(feature = "llmp_compression")]
+            let event_bytes = if *msg_flags & LLMP_FLAG_COMPRESSED == LLMP_FLAG_COMPRESSED {
+                compressed = compressor.decompress(msg)?;
+                &compressed
+            } else {
+                &*msg
+            };
+            let event: Event<I> = postcard::from_bytes(event_bytes)?;
+            match Self::handle_in_broker(monitor, client_id, &event)? {
+                BrokerEventResult::Forward => Ok(LlmpMsgHookResult::ForwardToClients),
+                BrokerEventResult::Handled => Ok(LlmpMsgHookResult::Handled),
+            }
+        } else {
+            Ok(LlmpMsgHookResult::ForwardToClients)
+        }
+    }
+
+    fn on_timeout(&mut self) -> Result<(), Error> {
+        self.monitor.display("Broker Heartbeat", ClientId(0));
+        Ok(())
+    }
+}
+
+impl<I, MT, SP> StdLlmpEventHook<I, MT, SP>
 where
     I: Input,
     SP: ShMemProvider + 'static,
     MT: Monitor,
 {
     /// Create an event broker from a raw broker.
-    pub fn new(llmp: llmp::LlmpBroker<SP>, monitor: MT) -> Result<Self, Error> {
+    pub fn new(monitor: MT) -> Result<Self, Error> {
         Ok(Self {
             monitor,
-            llmp,
             #[cfg(feature = "llmp_compression")]
             compressor: GzipCompressor::with_threshold(COMPRESS_THRESHOLD),
             phantom: PhantomData,
         })
     }
 
-    /// Create an LLMP broker on a port.
-    ///
-    /// The port must not be bound yet to have a broker.
-    #[cfg(feature = "std")]
-    pub fn on_port(shmem_provider: SP, monitor: MT, port: u16) -> Result<Self, Error> {
-        Ok(Self {
-            monitor,
-            llmp: llmp::LlmpBroker::create_attach_to_tcp(shmem_provider, port)?,
-            #[cfg(feature = "llmp_compression")]
-            compressor: GzipCompressor::with_threshold(COMPRESS_THRESHOLD),
-            phantom: PhantomData,
-        })
-    }
-
-    /// Exit the broker process cleanly after at least `n` clients attached and all of them disconnected again
-    pub fn set_exit_cleanly_after(&mut self, n_clients: NonZeroUsize) {
-        self.llmp.set_exit_cleanly_after(n_clients);
-    }
-
-    /// Connect to an LLMP broker on the given address
-    #[cfg(feature = "std")]
-    pub fn connect_b2b<A>(&mut self, addr: A) -> Result<(), Error>
-    where
-        A: ToSocketAddrs,
-    {
-        self.llmp.connect_b2b(addr)
-    }
-
-    /// Run forever in the broker
-    #[cfg(not(feature = "llmp_broker_timeouts"))]
-    pub fn broker_loop(&mut self) -> Result<(), Error> {
-        let monitor = &mut self.monitor;
-        #[cfg(feature = "llmp_compression")]
-        let compressor = &self.compressor;
-        self.llmp.loop_forever(
-            &mut |client_id, tag, _flags, msg| {
-                if tag == LLMP_TAG_EVENT_TO_BOTH {
-                    #[cfg(not(feature = "llmp_compression"))]
-                    let event_bytes = msg;
-                    #[cfg(feature = "llmp_compression")]
-                    let compressed;
-                    #[cfg(feature = "llmp_compression")]
-                    let event_bytes = if _flags & LLMP_FLAG_COMPRESSED == LLMP_FLAG_COMPRESSED {
-                        compressed = compressor.decompress(msg)?;
-                        &compressed
-                    } else {
-                        msg
-                    };
-                    let event: Event<I> = postcard::from_bytes(event_bytes)?;
-                    match Self::handle_in_broker(monitor, client_id, &event)? {
-                        BrokerEventResult::Forward => Ok(llmp::LlmpMsgHookResult::ForwardToClients),
-                        BrokerEventResult::Handled => Ok(llmp::LlmpMsgHookResult::Handled),
-                    }
-                } else {
-                    Ok(llmp::LlmpMsgHookResult::ForwardToClients)
-                }
-            },
-            Some(Duration::from_millis(5)),
-        );
-
-        #[cfg(all(feature = "std", feature = "llmp_debug"))]
-        println!("The last client quit. Exiting.");
-
-        Err(Error::shutting_down())
-    }
-
-    /// Run in the broker until all clients exit
-    #[cfg(feature = "llmp_broker_timeouts")]
-    pub fn broker_loop(&mut self) -> Result<(), Error> {
-        let monitor = &mut self.monitor;
-        #[cfg(feature = "llmp_compression")]
-        let compressor = &self.compressor;
-        self.llmp.loop_with_timeouts(
-            &mut |msg_or_timeout| {
-                if let Some((client_id, tag, _flags, msg)) = msg_or_timeout {
-                    if tag == LLMP_TAG_EVENT_TO_BOTH {
-                        #[cfg(not(feature = "llmp_compression"))]
-                        let event_bytes = msg;
-                        #[cfg(feature = "llmp_compression")]
-                        let compressed;
-                        #[cfg(feature = "llmp_compression")]
-                        let event_bytes = if _flags & LLMP_FLAG_COMPRESSED == LLMP_FLAG_COMPRESSED {
-                            compressed = compressor.decompress(msg)?;
-                            &compressed
-                        } else {
-                            msg
-                        };
-                        let event: Event<I> = postcard::from_bytes(event_bytes)?;
-                        match Self::handle_in_broker(monitor, client_id, &event)? {
-                            BrokerEventResult::Forward => {
-                                Ok(llmp::LlmpMsgHookResult::ForwardToClients)
-                            }
-                            BrokerEventResult::Handled => Ok(llmp::LlmpMsgHookResult::Handled),
-                        }
-                    } else {
-                        Ok(llmp::LlmpMsgHookResult::ForwardToClients)
-                    }
-                } else {
-                    monitor.display("Broker Heartbeat", ClientId(0));
-                    Ok(llmp::LlmpMsgHookResult::Handled)
-                }
-            },
-            Duration::from_secs(30),
-            Some(Duration::from_millis(5)),
-        );
-
-        #[cfg(feature = "llmp_debug")]
-        println!("The last client quit. Exiting.");
-
-        Err(Error::shutting_down())
-    }
-
     /// Handle arriving events in the broker
     #[allow(clippy::unnecessary_wraps)]
     fn handle_in_broker(

libafl/src/events/llmp/mod.rs

@@ -24,14 +24,14 @@ use crate::{
     Error, HasMetadata,
 };
 
-/// The llmp broker
-pub mod broker;
-pub use broker::*;
-
 /// The llmp event manager
 pub mod mgr;
 pub use mgr::*;
 
+/// The llmp hooks
+pub mod hooks;
+pub use hooks::*;
+
 /// The llmp restarting manager
 #[cfg(feature = "std")]
 pub mod restarting;

libafl/src/events/llmp/restarting.rs

@@ -8,7 +8,7 @@ use alloc::vec::Vec;
 use core::ptr::addr_of_mut;
 #[cfg(feature = "std")]
 use core::sync::atomic::{compiler_fence, Ordering};
-#[cfg(all(feature = "std", feature = "adaptive_serialization"))]
+#[cfg(feature = "std")]
 use core::time::Duration;
 use core::{marker::PhantomData, num::NonZeroUsize};
 #[cfg(feature = "std")]
@@ -24,11 +24,11 @@ use libafl_bolts::os::unix_signals::setup_signal_handler;
 use libafl_bolts::os::{fork, ForkResult};
 #[cfg(feature = "adaptive_serialization")]
 use libafl_bolts::tuples::{Handle, Handled};
+use libafl_bolts::{llmp::LlmpBroker, shmem::ShMemProvider, tuples::tuple_list};
 #[cfg(feature = "std")]
 use libafl_bolts::{
     llmp::LlmpConnection, os::CTRL_C_EXIT, shmem::StdShMemProvider, staterestore::StateRestorer,
 };
-use libafl_bolts::{shmem::ShMemProvider, tuples::tuple_list};
 use serde::{Deserialize, Serialize};
 #[cfg(feature = "std")]
 use typed_builder::TypedBuilder;
@@ -44,8 +44,8 @@ use crate::observers::TimeObserver;
 use crate::{
     events::{
         hooks::EventManagerHooksTuple, Event, EventConfig, EventFirer, EventManager,
-        EventManagerId, EventProcessor, EventRestarter, HasEventManagerId, LlmpEventBroker,
+        EventManagerId, EventProcessor, EventRestarter, HasEventManagerId, LlmpEventManager,
-        LlmpEventManager, LlmpShouldSaveState, ProgressReporter,
+        LlmpShouldSaveState, ProgressReporter, StdLlmpEventHook,
     },
     executors::{Executor, HasObservers},
     fuzzer::{Evaluator, EvaluatorObservers, ExecutionProcessor},
@@ -323,7 +323,7 @@ pub enum ManagerKind {
         /// The CPU core ID of this client
         cpu_core: Option<CoreId>,
     },
-    /// A [`crate::events::llmp::broker::LlmpEventBroker`], forwarding the packets of local clients.
+    /// An [`LlmpBroker`], forwarding the packets of local clients.
     Broker,
 }
 
@@ -344,8 +344,8 @@ pub fn setup_restarting_mgr_std<MT, S>(
     Error,
 >
 where
-    MT: Monitor + Clone,
+    MT: Monitor + Clone + 'static,
-    S: State + HasExecutions,
+    S: State + HasExecutions + 'static,
 {
     RestartingMgr::builder()
         .shmem_provider(StdShMemProvider::new()?)
@@ -375,8 +375,8 @@ pub fn setup_restarting_mgr_std<MT, S>(
     Error,
 >
 where
-    MT: Monitor + Clone,
+    MT: Monitor + Clone + 'static,
-    S: State + HasExecutions,
+    S: State + HasExecutions + 'static,
 {
     RestartingMgr::builder()
         .shmem_provider(StdShMemProvider::new()?)
@@ -448,8 +448,8 @@ impl<EMH, MT, S, SP> RestartingMgr<EMH, MT, S, SP>
 where
     EMH: EventManagerHooksTuple<S> + Copy + Clone,
     SP: ShMemProvider,
-    S: State + HasExecutions,
+    S: State + HasExecutions + 'static,
-    MT: Monitor + Clone,
+    MT: Monitor + Clone + 'static,
 {
     /// Launch the broker and the clients and fuzz
     pub fn launch(&mut self) -> Result<(Option<S>, LlmpRestartingEventManager<EMH, S, SP>), Error> {
@@ -457,18 +457,24 @@ where
         let (staterestorer, new_shmem_provider, core_id) = if std::env::var(_ENV_FUZZER_SENDER)
             .is_err()
         {
-            let broker_things = |mut broker: LlmpEventBroker<S::Input, MT, SP>,
+            let broker_things = |mut broker: LlmpBroker<_, SP>, remote_broker_addr| {
-                                 remote_broker_addr| {
                 if let Some(remote_broker_addr) = remote_broker_addr {
                     log::info!("B2b: Connecting to {:?}", &remote_broker_addr);
-                    broker.connect_b2b(remote_broker_addr)?;
+                    broker.state_mut().connect_b2b(remote_broker_addr)?;
                 };
 
                 if let Some(exit_cleanly_after) = self.exit_cleanly_after {
-                    broker.set_exit_cleanly_after(exit_cleanly_after);
+                    broker
+                        .state_mut()
+                        .set_exit_cleanly_after(exit_cleanly_after);
                 }
 
-                broker.broker_loop()
+                broker.loop_with_timeouts(Duration::from_secs(30), Some(Duration::from_millis(5)));
+
+                #[cfg(feature = "llmp_debug")]
+                log::info!("The last client quit. Exiting.");
+
+                Err(Error::shutting_down())
             };
             // We get here if we are on Unix, or we are a broker on Windows (or without forks).
             let (mgr, core_id) = match self.kind {
@@ -477,8 +483,7 @@ where
                         LlmpConnection::on_port(self.shmem_provider.clone(), self.broker_port)?;
                     match connection {
                         LlmpConnection::IsBroker { broker } => {
-                            let event_broker = LlmpEventBroker::<S::Input, MT, SP>::new(
+                            let llmp_hook = StdLlmpEventHook::<S::Input, MT, SP>::new(
-                                broker,
                                 self.monitor.take().unwrap(),
                             )?;
 
@@ -487,7 +492,10 @@ where
                                 "Doing broker things. Run this tool again to start fuzzing in a client."
                             );
 
-                            broker_things(event_broker, self.remote_broker_addr)?;
+                            broker_things(
+                                broker.add_hooks(tuple_list!(llmp_hook)),
+                                self.remote_broker_addr,
+                            )?;
 
                             return Err(Error::shutting_down());
                         }
@@ -511,13 +519,15 @@ where
                     }
                 }
                 ManagerKind::Broker => {
-                    let event_broker = LlmpEventBroker::<S::Input, MT, SP>::on_port(
+                    let llmp_hook = StdLlmpEventHook::new(self.monitor.take().unwrap())?;
+
+                    let broker = LlmpBroker::create_attach_to_tcp(
                         self.shmem_provider.clone(),
-                        self.monitor.take().unwrap(),
+                        tuple_list!(llmp_hook),
                         self.broker_port,
                     )?;
 
-                    broker_things(event_broker, self.remote_broker_addr)?;
+                    broker_things(broker, self.remote_broker_addr)?;
                     unreachable!("The broker may never return normally, only on errors or when shutting down.");
                 }
                 ManagerKind::Client { cpu_core } => {

libafl/src/events/tcp.rs

@@ -772,7 +772,7 @@ where
                 Ok(()) => {
                     self.tcp.set_nonblocking(false).expect("set to blocking");
                     let len = u32::from_le_bytes(len_buf);
-                    let mut buf = vec![0_u8; len as usize + 4_usize];
+                    let mut buf = vec![0_u8; 4_usize + len as usize];
                     self.tcp.read_exact(&mut buf)?;
 
                     let mut client_id_buf = [0_u8; 4];

libafl_bolts/examples/llmp_test/main.rs

@@ -5,15 +5,18 @@ extern crate alloc;
 
 #[cfg(not(target_os = "haiku"))]
 use core::time::Duration;
+use std::marker::PhantomData;
 #[cfg(all(feature = "std", not(target_os = "haiku")))]
 use std::{num::NonZeroUsize, thread, time};
 
+use libafl_bolts::{bolts_prelude::LlmpMsgHookResult, llmp::LlmpBrokerState};
 #[cfg(all(feature = "std", not(target_os = "haiku")))]
 use libafl_bolts::{
-    llmp::{self, Tag},
+    llmp::{self, Flags, LlmpHook, Tag},
     shmem::{ShMemProvider, StdShMemProvider},
     ClientId, Error, SimpleStderrLogger,
 };
+use tuple_list::tuple_list;
 
 #[cfg(all(feature = "std", not(target_os = "haiku")))]
 const _TAG_SIMPLE_U32_V1: Tag = Tag(0x5130_0321);
@@ -90,39 +93,68 @@ fn large_msg_loop(port: u16) -> Result<(), Box<dyn std::error::Error>> {
     }
 }
 
-#[allow(clippy::unnecessary_wraps)]
+pub struct LlmpExampleHook<SP> {
+    phantom: PhantomData<SP>,
+}
+
+impl<SP> LlmpExampleHook<SP> {
+    #[must_use]
+    pub fn new() -> Self {
+        Self {
+            phantom: PhantomData,
+        }
+    }
+}
+
+impl<SP> Default for LlmpExampleHook<SP> {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
 #[cfg(all(feature = "std", not(target_os = "haiku")))]
-fn broker_message_hook(
+impl<SP> LlmpHook<SP> for LlmpExampleHook<SP>
-    msg_or_timeout: Option<(ClientId, llmp::Tag, llmp::Flags, &[u8])>,
+where
-) -> Result<llmp::LlmpMsgHookResult, Error> {
+    SP: ShMemProvider + 'static,
-    let Some((client_id, tag, _flags, message)) = msg_or_timeout else {
+{
+    fn on_new_message(
+        &mut self,
+        _llmp_broker_state: &mut LlmpBrokerState<SP>,
+        client_id: ClientId,
+        msg_tag: &mut Tag,
+        _msg_flags: &mut Flags,
+        msg: &mut [u8],
+    ) -> Result<LlmpMsgHookResult, Error> {
+        match *msg_tag {
+            _TAG_SIMPLE_U32_V1 => {
+                println!(
+                    "Client {:?} sent message: {:?}",
+                    client_id,
+                    u32::from_le_bytes(msg.try_into()?)
+                );
+                Ok(LlmpMsgHookResult::ForwardToClients)
+            }
+            _TAG_MATH_RESULT_V1 => {
+                println!(
+                    "Adder Client has this current result: {:?}",
+                    u32::from_le_bytes(msg.try_into()?)
+                );
+                Ok(LlmpMsgHookResult::Handled)
+            }
+            _ => {
+                println!("Unknown message id received: {msg_tag:?}");
+                Ok(LlmpMsgHookResult::ForwardToClients)
+            }
+        }
+    }
+
+    fn on_timeout(&mut self) -> Result<(), Error> {
         println!(
             "No client did anything for {} seconds..",
             BROKER_TIMEOUT.as_secs()
         );
-        return Ok(llmp::LlmpMsgHookResult::Handled);
-    };
 
-    match tag {
+        Ok(())
-        _TAG_SIMPLE_U32_V1 => {
-            println!(
-                "Client {:?} sent message: {:?}",
-                client_id,
-                u32::from_le_bytes(message.try_into()?)
-            );
-            Ok(llmp::LlmpMsgHookResult::ForwardToClients)
-        }
-        _TAG_MATH_RESULT_V1 => {
-            println!(
-                "Adder Client has this current result: {:?}",
-                u32::from_le_bytes(message.try_into()?)
-            );
-            Ok(llmp::LlmpMsgHookResult::Handled)
-        }
-        _ => {
-            println!("Unknown message id received: {tag:?}");
-            Ok(llmp::LlmpMsgHookResult::ForwardToClients)
-        }
     }
 }
 
@@ -154,26 +186,26 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
 
     match mode.as_str() {
         "broker" => {
-            let mut broker = llmp::LlmpBroker::new(StdShMemProvider::new()?)?;
+            let mut broker = llmp::LlmpBroker::new(
-            broker.launch_tcp_listener_on(port)?;
+                StdShMemProvider::new()?,
+                tuple_list!(LlmpExampleHook::new()),
+            )?;
+            broker.state_mut().launch_tcp_listener_on(port)?;
             // Exit when we got at least _n_ nodes, and all of them quit.
-            broker.set_exit_cleanly_after(NonZeroUsize::new(1_usize).unwrap());
+            broker
-            broker.loop_with_timeouts(
+                .state_mut()
-                &mut broker_message_hook,
+                .set_exit_cleanly_after(NonZeroUsize::new(1_usize).unwrap());
-                BROKER_TIMEOUT,
+            broker.loop_with_timeouts(BROKER_TIMEOUT, Some(SLEEP_BETWEEN_FORWARDS));
-                Some(SLEEP_BETWEEN_FORWARDS),
-            );
         }
         "b2b" => {
-            let mut broker = llmp::LlmpBroker::new(StdShMemProvider::new()?)?;
+            let mut broker = llmp::LlmpBroker::new(
-            broker.launch_tcp_listener_on(b2b_port)?;
+                StdShMemProvider::new()?,
+                tuple_list!(LlmpExampleHook::new()),
+            )?;
+            broker.state_mut().launch_tcp_listener_on(b2b_port)?;
             // connect back to the main broker.
-            broker.connect_b2b(("127.0.0.1", port))?;
+            broker.state_mut().connect_b2b(("127.0.0.1", port))?;
-            broker.loop_with_timeouts(
+            broker.loop_with_timeouts(BROKER_TIMEOUT, Some(SLEEP_BETWEEN_FORWARDS));
-                &mut broker_message_hook,
-                BROKER_TIMEOUT,
-                Some(SLEEP_BETWEEN_FORWARDS),
-            );
         }
         "ctr" => {
             let mut client =

libafl_bolts/src/core_affinity.rs

@@ -235,7 +235,7 @@ fn set_for_current_helper(core_id: CoreId) -> Result<(), Error> {
 ))]
 mod linux {
     use alloc::{string::ToString, vec::Vec};
-    use std::mem;
+    use core::mem::{size_of, zeroed};
 
     #[cfg(not(target_os = "freebsd"))]
     use libc::cpu_set_t;
@@ -276,7 +276,7 @@ mod linux {
         let result = unsafe {
             sched_setaffinity(
                 0, // Defaults to current thread
-                mem::size_of::<cpu_set_t>(),
+                size_of::<cpu_set_t>(),
                 &set,
             )
         };
@@ -295,7 +295,7 @@ mod linux {
         let result = unsafe {
             sched_getaffinity(
                 0, // Defaults to current thread
-                mem::size_of::<cpu_set_t>(),
+                size_of::<cpu_set_t>(),
                 &mut set,
             )
         };
@@ -310,7 +310,7 @@ mod linux {
     }
 
     fn new_cpu_set() -> cpu_set_t {
-        unsafe { mem::zeroed::<cpu_set_t>() }
+        unsafe { zeroed::<cpu_set_t>() }
     }
 
     #[cfg(test)]

libafl_bolts/src/os/pipes.rs

@@ -1,18 +1,20 @@
 //! Unix `pipe` wrapper for `LibAFL`
-use alloc::rc::Rc;
-use core::{borrow::Borrow, cell::RefCell};
 #[cfg(feature = "std")]
 use std::{
+    borrow::Borrow,
+    cell::RefCell,
     io::{self, ErrorKind, Read, Write},
     os::{
         fd::{AsFd, AsRawFd, OwnedFd},
         unix::io::RawFd,
     },
+    rc::Rc,
 };
 
 #[cfg(feature = "std")]
 use nix::unistd::{pipe, read, write};
 
+#[cfg(feature = "std")]
 use crate::Error;
 
 /// A unix pipe wrapper for `LibAFL`

libafl_bolts/src/os/unix_signals.rs

@@ -1,6 +1,8 @@
 //! Signal handling for unix
 #[cfg(feature = "alloc")]
 use alloc::vec::Vec;
+#[cfg(all(target_vendor = "apple", target_arch = "aarch64"))]
+use core::mem::size_of;
 #[cfg(feature = "alloc")]
 use core::{
     cell::UnsafeCell,
@@ -160,7 +162,7 @@ pub struct arm_thread_state64 {
 //#[repr(align(16))]
 pub struct arm_neon_state64 {
     /// opaque
-    pub opaque: [u8; (32 * 16) + (2 * mem::size_of::<u32>())],
+    pub opaque: [u8; (32 * 16) + (2 * size_of::<u32>())],
 }
 
 /// ```c

libafl_bolts/src/shmem.rs

@@ -7,7 +7,7 @@ use alloc::{rc::Rc, string::ToString, vec::Vec};
 use core::{cell::RefCell, fmt, fmt::Display, mem::ManuallyDrop};
 use core::{
     fmt::Debug,
-    mem,
+    mem::size_of,
     ops::{Deref, DerefMut},
 };
 #[cfg(feature = "std")]
@@ -202,7 +202,7 @@ pub trait ShMem: Sized + Debug + Clone + DerefMut<Target = [u8]> {
     /// Convert to a ptr of a given type, checking the size.
     /// If the map is too small, returns `None`
     fn as_ptr_of<T: Sized>(&self) -> Option<*const T> {
-        if self.len() >= mem::size_of::<T>() {
+        if self.len() >= size_of::<T>() {
             Some(self.as_ptr() as *const T)
         } else {
             None
@@ -212,7 +212,7 @@ pub trait ShMem: Sized + Debug + Clone + DerefMut<Target = [u8]> {
     /// Convert to a mut ptr of a given type, checking the size.
     /// If the map is too small, returns `None`
     fn as_mut_ptr_of<T: Sized>(&mut self) -> Option<*mut T> {
-        if self.len() >= mem::size_of::<T>() {
+        if self.len() >= size_of::<T>() {
             Some(self.as_mut_ptr() as *mut T)
         } else {
             None
@@ -267,7 +267,7 @@ pub trait ShMemProvider: Clone + Default + Debug {
 
     /// Create a new shared memory mapping to hold an object of the given type, and initializes it with the given value.
     fn uninit_on_shmem<T: Sized + 'static>(&mut self) -> Result<Self::ShMem, Error> {
-        self.new_shmem(mem::size_of::<T>())
+        self.new_shmem(size_of::<T>())
     }
 
     /// Get a mapping given a description

libafl_libfuzzer/libafl_libfuzzer_runtime/src/fuzz.rs

@@ -139,7 +139,7 @@ fn fuzz_many_forking<M>(
     monitor: M,
 ) -> Result<(), Error>
 where
-    M: Monitor + Clone + Debug,
+    M: Monitor + Clone + Debug + 'static,
 {
     destroy_output_fds(options);
     let broker_port = std::env::var(PORT_PROVIDER_VAR)

libafl_targets/src/cmps/mod.rs

@@ -8,7 +8,8 @@ use alloc::{alloc::alloc_zeroed, boxed::Box, vec::Vec};
 use core::{
     alloc::Layout,
     fmt::{self, Debug, Formatter},
-    mem, ptr, slice,
+    mem::{size_of, zeroed},
+    ptr, slice,
 };
 
 use libafl::{
@@ -30,11 +31,11 @@ pub const CMPLOG_RTN_LEN: usize = 32;
 
 /// The hight of a cmplog routine map
 pub const CMPLOG_MAP_RTN_H: usize =
-    (CMPLOG_MAP_H * mem::size_of::<CmpLogInstruction>()) / mem::size_of::<CmpLogRoutine>();
+    (CMPLOG_MAP_H * size_of::<CmpLogInstruction>()) / size_of::<CmpLogRoutine>();
 
 /// The height of extended rountine map
 pub const CMPLOG_MAP_RTN_EXTENDED_H: usize =
-    CMPLOG_MAP_H * mem::size_of::<AFLppCmpLogOperands>() / mem::size_of::<AFLppCmpLogFnOperands>();
+    CMPLOG_MAP_H * size_of::<AFLppCmpLogOperands>() / size_of::<AFLppCmpLogFnOperands>();
 
 /// `CmpLog` instruction kind
 pub const CMPLOG_KIND_INS: u8 = 0;
@@ -318,7 +319,7 @@ pub struct CmpLogMap {
 
 impl Default for CmpLogMap {
     fn default() -> Self {
-        unsafe { mem::zeroed() }
+        unsafe { zeroed() }
     }
 }
 
@@ -474,9 +475,8 @@ impl Serialize for AFLppCmpLogMap {
     where
         S: Serializer,
     {
-        let slice = unsafe {
+        let slice =
-            slice::from_raw_parts(ptr::from_ref(self) as *const u8, mem::size_of::<Self>())
+            unsafe { slice::from_raw_parts(ptr::from_ref(self) as *const u8, size_of::<Self>()) };
-        };
         serializer.serialize_bytes(slice)
     }
 }

libafl_targets/src/sancov_8bit.rs

@@ -64,6 +64,7 @@ mod observers {
         fmt::Debug,
         hash::{Hash, Hasher},
         iter::Flatten,
+        mem::size_of,
         ptr::{addr_of, addr_of_mut},
         slice::{from_raw_parts, Iter, IterMut},
     };
@@ -161,7 +162,7 @@ mod observers {
             for map in unsafe { &*addr_of!(COUNTERS_MAPS) } {
                 let slice = map.as_slice();
                 let ptr = slice.as_ptr();
-                let map_size = slice.len() / core::mem::size_of::<u8>();
+                let map_size = slice.len() / size_of::<u8>();
                 unsafe {
                     hasher.write(from_raw_parts(ptr, map_size));
                 }

libafl_targets/src/sancov_pcguard.rs

@@ -3,7 +3,7 @@
 #[rustversion::nightly]
 #[cfg(feature = "sancov_ngram4")]
 use core::simd::num::SimdUint;
-use core::{mem, ptr, slice};
+use core::{mem::align_of, ptr, slice};
 
 #[cfg(any(feature = "sancov_ngram4", feature = "sancov_ctx"))]
 use libafl::executors::{hooks::ExecutorHook, HasObservers};
@@ -346,7 +346,7 @@ pub fn sanitizer_cov_pc_table() -> Option<&'static [PcTableEntry]> {
             "PC Table size is not evens - start: {PCS_BEG:x?} end: {PCS_END:x?}"
         );
         assert_eq!(
-            (PCS_BEG as usize) % mem::align_of::<PcTableEntry>(),
+            (PCS_BEG as usize) % align_of::<PcTableEntry>(),
             0,
             "Unaligned PC Table - start: {PCS_BEG:x?} end: {PCS_END:x?}"
         );