: lints

hyperactor/src/channel.rs

@@ -906,7 +906,6 @@ mod tests {
     use super::*;
     use crate::clock::Clock;
     use crate::clock::RealClock;
-    use crate::config;
 
     #[test]
     fn test_channel_addr() {

hyperactor/src/mailbox/undeliverable.rs

@@ -13,10 +13,8 @@ use serde::Serialize;
 use thiserror::Error;
 
 use crate as hyperactor; // for macros
-use crate::ActorId;
 use crate::Message;
 use crate::Named;
-use crate::PortId;
 use crate::actor::ActorStatus;
 use crate::id;
 use crate::mailbox::DeliveryError;

hyperactor_macros/src/lib.rs

@@ -1636,10 +1636,7 @@ impl syn::parse::Parse for BehaviorInput {
 /// ```
 #[proc_macro]
 pub fn behavior(input: TokenStream) -> TokenStream {
-    let BehaviorInput {
-        behavior: behavior,
-        handlers,
-    } = parse_macro_input!(input as BehaviorInput);
+    let BehaviorInput { behavior, handlers } = parse_macro_input!(input as BehaviorInput);
     let tys = HandlerSpec::add_indexed(handlers);
 
     let expanded = quote! {

hyperactor_mesh/Cargo.toml

@@ -65,7 +65,6 @@ libc = "0.2.139"
 mockall = "0.13.1"
 ndslice = { version = "0.0.0", path = "../ndslice" }
 nix = { version = "0.30.1", features = ["dir", "event", "hostname", "inotify", "ioctl", "mman", "mount", "net", "poll", "ptrace", "reboot", "resource", "sched", "signal", "term", "time", "user", "zerocopy"] }
-notify = "5"
 opentelemetry = "0.29"
 pin-project = "1.1.10"
 preempt_rwlock = { version = "0.0.0", path = "../preempt_rwlock" }

hyperactor_mesh/examples/dining_philosophers.rs

@@ -23,7 +23,6 @@ use hyperactor::Named;
 use hyperactor::PortRef;
 use hyperactor::Unbind;
 use hyperactor::context;
-use hyperactor_mesh::bootstrap::BootstrapCommand;
 use hyperactor_mesh::comm::multicast::CastInfo;
 use hyperactor_mesh::extent;
 use hyperactor_mesh::proc_mesh::global_root_client;

hyperactor_mesh/examples/test_bench.rs

@@ -22,6 +22,8 @@ use hyperactor::Handler;
 use hyperactor::Named;
 use hyperactor::PortRef;
 use hyperactor::Unbind;
+use hyperactor::clock::Clock;
+use hyperactor::clock::RealClock;
 use hyperactor_mesh::bootstrap::BootstrapCommand;
 use hyperactor_mesh::comm::multicast::CastInfo;
 use hyperactor_mesh::proc_mesh::global_root_client;
@@ -31,7 +33,6 @@ use ndslice::ViewExt;
 use ndslice::extent;
 use serde::Deserialize;
 use serde::Serialize;
-use tokio::time::Instant;
 
 #[derive(Actor, Default, Debug)]
 #[hyperactor::export(
@@ -84,7 +85,7 @@ async fn main() {
     loop {
         let mut received = HashSet::new();
         let (port, mut rx) = instance.open_port();
-        let begin = Instant::now();
+        let begin = RealClock.now();
         actor_mesh
             .cast(instance, TestMessage::Ping(port.bind()))
             .unwrap();
@@ -93,6 +94,6 @@ async fn main() {
         }
 
         eprintln!("ping {}ms", begin.elapsed().as_millis());
-        tokio::time::sleep(Duration::from_secs(1)).await;
+        RealClock.sleep(Duration::from_secs(1)).await;
     }
 }

hyperactor_mesh/src/actor_mesh.rs

@@ -984,7 +984,7 @@ mod tests {
                     .await
                     .unwrap();
 
-                let instance = crate::v1::testing::instance().await;
+                let instance = $crate::v1::testing::instance().await;
                 let proc_mesh = ProcMesh::allocate(alloc).await.unwrap();
                 let (undeliverable_tx, _undeliverable_rx) = proc_mesh.client().open_port();
                 let params = PingPongActorParams::new(Some(undeliverable_tx.bind()), None);
@@ -1030,7 +1030,7 @@ mod tests {
                     .await
                     .unwrap();
 
-                let instance = crate::v1::testing::instance().await;
+                let instance = $crate::v1::testing::instance().await;
                 let proc_mesh = ProcMesh::allocate(alloc).await.unwrap();
                 let actor_mesh: RootActorMesh<TestActor> = proc_mesh.spawn(&instance, "echo", &()).await.unwrap();
                 let dont_simulate_error = true;
@@ -1074,7 +1074,7 @@ mod tests {
                     .await
                     .unwrap();
 
-                let instance = crate::v1::testing::instance().await;
+                let instance = $crate::v1::testing::instance().await;
                 let proc_mesh = ProcMesh::allocate(alloc).await.unwrap();
                 let actor_mesh: RootActorMesh<TestActor> = proc_mesh.spawn(&instance, "echo", &()).await.unwrap();
 
@@ -1096,7 +1096,7 @@ mod tests {
             #[tokio::test]
             async fn test_inter_proc_mesh_comms() {
                 let mut meshes = Vec::new();
-                let instance = crate::v1::testing::instance().await;
+                let instance = $crate::v1::testing::instance().await;
                 for _ in 0..2 {
                     let alloc = $allocator
                         .allocate(AllocSpec {
@@ -1157,7 +1157,7 @@ mod tests {
                     .await
                     .unwrap();
 
-                let instance = crate::v1::testing::instance().await;
+                let instance = $crate::v1::testing::instance().await;
                 let mut proc_mesh = ProcMesh::allocate(alloc).await.unwrap();
 
                 let (tx, mut rx) = hyperactor::mailbox::open_port(proc_mesh.client());
@@ -1220,7 +1220,7 @@ mod tests {
                     .await
                     .unwrap();
 
-                let instance = crate::v1::testing::instance().await;
+                let instance = $crate::v1::testing::instance().await;
                 let mesh = ProcMesh::allocate(alloc).await.unwrap();
                 let (reply_port_handle, mut reply_port_receiver) = mesh.client().open_port::<usize>();
                 let reply_port = reply_port_handle.bind();

hyperactor_mesh/src/bootstrap.rs

@@ -257,11 +257,13 @@ pub enum Bootstrap {
 impl Bootstrap {
     /// Serialize the mode into a environment-variable-safe string by
     /// base64-encoding its JSON representation.
+    #[allow(clippy::result_large_err)]
     fn to_env_safe_string(&self) -> v1::Result<String> {
         Ok(BASE64_STANDARD.encode(serde_json::to_string(&self)?))
     }
 
     /// Deserialize the mode from the representation returned by [`to_env_safe_string`].
+    #[allow(clippy::result_large_err)]
     fn from_env_safe_string(str: &str) -> v1::Result<Self> {
         let data = BASE64_STANDARD.decode(str)?;
         let data = std::str::from_utf8(&data)?;

hyperactor_mesh/src/logging.rs

@@ -47,7 +47,6 @@ use hyperactor::data::Serialized;
 use hyperactor::declare_attrs;
 use hyperactor_telemetry::env;
 use hyperactor_telemetry::log_file_path;
-use notify::Watcher;
 use serde::Deserialize;
 use serde::Serialize;
 use tokio::io;
@@ -1657,12 +1656,14 @@ mod tests {
             "Expected deserialization to fail with invalid UTF-8 bytes"
         );
     }
+    #[allow(dead_code)]
     struct MockLogSender {
         log_sender: mpsc::UnboundedSender<(OutputTarget, String)>, // (output_target, content)
         flush_called: Arc<Mutex<bool>>,                            // Track if flush was called
     }
 
     impl MockLogSender {
+        #[allow(dead_code)]
         fn new(log_sender: mpsc::UnboundedSender<(OutputTarget, String)>) -> Self {
             Self {
                 log_sender,

hyperactor_mesh/src/logging/line_prefixing_writer.rs

@@ -27,6 +27,7 @@ where
     W: io::AsyncWrite + Unpin,
 {
     /// Creates a new LinePrefixingWriter that will prefix each line with "[<local_rank>] ".
+    #[allow(dead_code)]
     pub fn new(local_rank: usize, inner: W) -> Self {
         let prefix = format!("[{}] ", local_rank).into_bytes();
         Self {

hyperactor_mesh/src/proc_mesh.rs

@@ -697,6 +697,7 @@ impl ProcMesh {
                     .collect(),
                 actor_event_router: actor_event_router.clone(),
             }),
+            #[allow(clippy::todo)]
             ProcMeshKind::V1(_proc_mesh) => todo!(),
         }
     }

hyperactor_mesh/src/v1.rs

@@ -124,13 +124,13 @@ pub enum Error {
 
     #[error(
         "error spawning proc mesh: statuses: {}",
-        RankedValues::invert(&*.statuses)
+        RankedValues::invert(statuses)
     )]
     ProcSpawnError { statuses: RankedValues<Status> },
 
     #[error(
         "error spawning actor mesh: statuses: {}",
-        RankedValues::invert(&*.statuses)
+        RankedValues::invert(statuses)
     )]
     ActorSpawnError { statuses: RankedValues<Status> },
 

hyperactor_mesh/src/v1/actor_mesh.rs

@@ -158,6 +158,7 @@ pub struct ActorMeshRef<A: Referable> {
 
 impl<A: Referable> ActorMeshRef<A> {
     /// Cast a message to all the actors in this mesh
+    #[allow(clippy::result_large_err)]
     pub fn cast<M>(&self, cx: &impl context::Actor, message: M) -> v1::Result<()>
     where
         A: RemoteHandles<M> + RemoteHandles<IndexedErasedUnbound<M>>,
@@ -170,6 +171,7 @@ impl<A: Referable> ActorMeshRef<A> {
     /// This should *only* be used for temporary support for selections in the tensor
     /// engine. If you use this for anything else, you will be fired (you too, OSS
     /// contributor).
+    #[allow(clippy::result_large_err)]
     pub(crate) fn cast_for_tensor_engine_only_do_not_use<M>(
         &self,
         cx: &impl context::Actor,
@@ -183,6 +185,7 @@ impl<A: Referable> ActorMeshRef<A> {
         self.cast_with_selection(cx, sel, message)
     }
 
+    #[allow(clippy::result_large_err)]
     fn cast_with_selection<M>(
         &self,
         cx: &impl context::Actor,
@@ -226,6 +229,7 @@ impl<A: Referable> ActorMeshRef<A> {
         }
     }
 
+    #[allow(clippy::result_large_err)]
     fn cast_v0<M>(
         &self,
         cx: &impl context::Actor,
@@ -266,6 +270,7 @@ impl<A: Referable> ActorMeshRef<A> {
         }
     }
 
+    #[allow(clippy::result_large_err)]
     pub async fn actor_states(
         &self,
         cx: &impl context::Actor,

hyperactor_mesh/src/v1/host_mesh.rs

@@ -606,6 +606,7 @@ pub struct HostMeshRef {
 impl HostMeshRef {
     /// Create a new (raw) HostMeshRef from the provided region and associated
     /// ranks, which must match in cardinality.
+    #[allow(clippy::result_large_err)]
     fn new(name: Name, region: Region, ranks: Vec<HostRef>) -> v1::Result<Self> {
         if region.num_ranks() != ranks.len() {
             return Err(v1::Error::InvalidRankCardinality {
@@ -635,6 +636,7 @@ impl HostMeshRef {
     ///
     /// Currently, spawn issues direct calls to each host agent. This will be fixed by
     /// maintaining a comm actor on the host service procs themselves.
+    #[allow(clippy::result_large_err)]
     pub async fn spawn(
         &self,
         cx: &impl context::Actor,
@@ -873,6 +875,7 @@ impl HostMeshRef {
 
     /// Get the state of all procs with Name in this host mesh.
     /// The procs iterator must be in rank order.
+    #[allow(clippy::result_large_err)]
     pub(crate) async fn proc_states(
         &self,
         cx: &impl context::Actor,

hyperactor_mesh/src/v1/proc_mesh.rs

@@ -187,8 +187,10 @@ impl ProcRef {
 /// A mesh of processes.
 #[derive(Debug)]
 pub struct ProcMesh {
+    #[allow(dead_code)]
     name: Name,
     allocation: ProcMeshAllocation,
+    #[allow(dead_code)]
     comm_actor_name: Option<Name>,
     current_ref: ProcMeshRef,
 }
@@ -430,6 +432,7 @@ impl ProcMesh {
     }
 
     /// Detach the proc mesh from the lifetime of `self`, and return its reference.
+    #[allow(dead_code)]
     pub(crate) fn detach(self) -> ProcMeshRef {
         // This also keeps the ProcMeshAllocation::Allocated alloc task alive.
         self.current_ref
@@ -495,7 +498,7 @@ enum ProcMeshAllocation {
 impl ProcMeshAllocation {
     fn extent(&self) -> &Extent {
         match self {
-            ProcMeshAllocation::Allocated { extent, .. } => &extent,
+            ProcMeshAllocation::Allocated { extent, .. } => extent,
             ProcMeshAllocation::Owned { extent, .. } => extent,
         }
     }
@@ -562,6 +565,7 @@ pub struct ProcMeshRef {
 
 impl ProcMeshRef {
     /// Create a new ProcMeshRef from the given name, region, ranks, and so on.
+    #[allow(clippy::result_large_err)]
     fn new(
         name: Name,
         region: Region,

hyperactor_mesh/src/v1/testactor.rs

@@ -29,7 +29,6 @@ use hyperactor::Named;
 use hyperactor::PortRef;
 use hyperactor::RefClient;
 use hyperactor::Unbind;
-use hyperactor::attrs::Attrs;
 #[cfg(test)]
 use hyperactor::clock::Clock as _;
 #[cfg(test)]
@@ -240,7 +239,7 @@ pub struct SetConfigAttrs(pub Vec<u8>);
 impl Handler<SetConfigAttrs> for TestActor {
     async fn handle(
         &mut self,
-        cx: &Context<Self>,
+        _cx: &Context<Self>,
         SetConfigAttrs(attrs): SetConfigAttrs,
     ) -> Result<(), anyhow::Error> {
         let attrs = bincode::deserialize(&attrs)?;

hyperactor_mesh/src/v1/value_mesh.rs

@@ -10,7 +10,6 @@ use std::cmp::Ordering;
 use std::collections::HashMap;
 use std::collections::hash_map::Entry;
 use std::hash::Hash;
-use std::hash::Hasher;
 use std::marker::PhantomData;
 use std::mem;
 use std::mem::MaybeUninit;
@@ -111,6 +110,7 @@ impl TryFrom<Run> for (Range<usize>, u32) {
     /// exceeds the platform's addressable range. This ensures safe
     /// round-tripping between the serialized wire format and native
     /// representation.
+    #[allow(clippy::result_large_err)]
     fn try_from(r: Run) -> Result<Self, Self::Error> {
         let start = usize::try_from(r.start).map_err(|_| "run.start too large")?;
         let end = usize::try_from(r.end).map_err(|_| "run.end too large")?;
@@ -203,6 +203,7 @@ impl<T> ValueMesh<T> {
     /// Returns [`Error::InvalidRankCardinality`] if `ranks.len() !=
     /// region.num_ranks()`.
     /// ```
+    #[allow(clippy::result_large_err)]
     pub(crate) fn new(region: Region, ranks: Vec<T>) -> crate::v1::Result<Self> {
         let (actual, expected) = (ranks.len(), region.num_ranks());
         if actual != expected {
@@ -275,6 +276,7 @@ impl<T: Eq + Hash> ValueMesh<T> {
     /// - Unspecified ranks are filled with `default`.
     /// - Result is stored in RLE form; no dense `Vec<T>` is
     ///   materialized.
+    #[allow(clippy::result_large_err)]
     pub fn from_ranges_with_default(
         region: Region,
         default: T,
@@ -410,6 +412,7 @@ impl<T: Eq + Hash> ValueMesh<T> {
     /// let mesh = ValueMesh::from_dense(region, vec![1, 1, 2, 2, 3]).unwrap();
     /// // Internally compressed to three runs: [1, 1], [2, 2], [3]
     /// ```
+    #[allow(clippy::result_large_err)]
     pub fn from_dense(region: Region, values: Vec<T>) -> crate::v1::Result<Self> {
         let mut vm = Self::new(region, values)?;
         vm.compress_adjacent_in_place();
@@ -1280,6 +1283,7 @@ mod tests {
     }
 
     // Indexed collector naïve implementation (for reference).
+    #[allow(clippy::result_large_err)]
     fn build_value_mesh_indexed<T>(
         region: Region,
         pairs: impl IntoIterator<Item = (usize, T)>,

monarch_extension/src/client.rs

@@ -578,7 +578,7 @@ impl ClientActor {
     }
 
     #[staticmethod]
-    fn new_with_parent(proc: &PyProc, parent: &PyActorId) -> PyResult<Self> {
+    fn new_with_parent(_proc: &PyProc, _parent: &PyActorId) -> PyResult<Self> {
         // XXX:
         unimplemented!("this is not a valid thing to do!");
         // Ok(Self {

monarch_extension/src/mesh_controller.rs

@@ -30,7 +30,6 @@ use hyperactor::Handler;
 use hyperactor::Instance;
 use hyperactor::PortRef;
 use hyperactor::ProcId;
-use hyperactor::actor::ActorStatus;
 use hyperactor::context;
 use hyperactor::mailbox::MailboxSenderError;
 use hyperactor_mesh::Mesh;
@@ -69,7 +68,6 @@ use monarch_tensor_worker::WorkerActor;
 use ndslice::Slice;
 use ndslice::ViewExt;
 use ndslice::selection::ReifySlice;
-use pyo3::exceptions::PyRuntimeError;
 use pyo3::exceptions::PyValueError;
 use pyo3::prelude::*;
 use tokio::sync::Mutex;