Add worker_find_logging

Author: palfrey

.github/workflows/native-cargo.yaml

@@ -46,3 +46,7 @@ jobs:
 
       - name: Test on ${{ runner.os }}
         run: cargo test --all --profile=smol
+
+      # Not a default target, but need to make sure we don't actually break it
+      - name: Test worker_find_logging
+        run: cargo build --features worker_find_logging --all-targets

BUILD.bazel

@@ -14,6 +14,8 @@ rust_binary(
     srcs = [
         "src/bin/nativelink.rs",
     ],
+    # Enable this to get extra debug about workers that are not being used by the CAS
+    # crate_features = ["worker_find_logging"],
     deps = [
         "//nativelink-config",
         "//nativelink-error",

Cargo.toml

@@ -25,6 +25,14 @@ name = "nativelink"
 [features]
 nix = ["nativelink-worker/nix"]
 
+# Enable this to get extra debug about workers that are not being used by the CAS
+# for some reason. We don't enable this by default, as it's part of a hot path in
+# the scheduling system, and also that a worker not matching isn't necessarily bad.
+worker_find_logging = [
+  "nativelink-scheduler/worker_find_logging",
+  "nativelink-util/worker_find_logging",
+]
+
 [dependencies]
 nativelink-config = { path = "nativelink-config" }
 nativelink-error = { path = "nativelink-error" }

flake.nix

@@ -161,6 +161,8 @@
           (craneLibFor p).buildPackage ((commonArgsFor p)
             // {
               cargoArtifacts = cargoArtifactsFor p;
+              # Enable this for debugging worker scheduler issues
+              # cargoExtraArgs = "--features worker_find_logging";
             });
 
         nativeTargetPkgs =

nativelink-scheduler/Cargo.toml

@@ -5,6 +5,9 @@ edition = "2024"
 name = "nativelink-scheduler"
 version = "0.7.0"
 
+[features]
+worker_find_logging = ["nativelink-util/worker_find_logging"]
+
 [dependencies]
 nativelink-config = { path = "../nativelink-config" }
 nativelink-error = { path = "../nativelink-error" }

nativelink-scheduler/src/api_worker_scheduler.rs

@@ -33,6 +33,8 @@ use nativelink_util::task::JoinHandleDropGuard;
 use tokio::sync::Notify;
 use tokio::sync::mpsc::{self, UnboundedSender};
 use tonic::async_trait;
+#[cfg(feature = "worker_find_logging")]
+use tracing::info;
 use tracing::{error, warn};
 
 use crate::platform_property_manager::PlatformPropertyManager;
@@ -191,21 +193,46 @@ impl ApiWorkerSchedulerImpl {
         Ok(())
     }
 
+    #[cfg_attr(not(feature = "worker_find_logging"), allow(unused_variables))]
+    fn inner_worker_checker(
+        (worker_id, w): &(&WorkerId, &Worker),
+        platform_properties: &PlatformProperties,
+    ) -> bool {
+        #[cfg(feature = "worker_find_logging")]
+        {
+            if !w.can_accept_work() {
+                info!(
+                    "Worker {worker_id} cannot accept work because is_paused: {}, is_draining: {}",
+                    w.is_paused, w.is_draining
+                );
+                return false;
+            }
+            if !platform_properties.is_satisfied_by(&w.platform_properties) {
+                info!("Worker {worker_id} properties are insufficient");
+                return false;
+            }
+            return true;
+        }
+        #[cfg(not(feature = "worker_find_logging"))]
+        {
+            w.can_accept_work() && platform_properties.is_satisfied_by(&w.platform_properties)
+        }
+    }
+
     fn inner_find_worker_for_action(
         &self,
         platform_properties: &PlatformProperties,
     ) -> Option<WorkerId> {
         let mut workers_iter = self.workers.iter();
-        let workers_iter = match self.allocation_strategy {
-            // Use rfind to get the least recently used that satisfies the properties.
-            WorkerAllocationStrategy::LeastRecentlyUsed => workers_iter.rfind(|(_, w)| {
-                w.can_accept_work() && platform_properties.is_satisfied_by(&w.platform_properties)
-            }),
-            // Use find to get the most recently used that satisfies the properties.
-            WorkerAllocationStrategy::MostRecentlyUsed => workers_iter.find(|(_, w)| {
-                w.can_accept_work() && platform_properties.is_satisfied_by(&w.platform_properties)
-            }),
-        };
+        let workers_iter =
+            match self.allocation_strategy {
+                // Use rfind to get the least recently used that satisfies the properties.
+                WorkerAllocationStrategy::LeastRecentlyUsed => workers_iter
+                    .rfind(|worker| Self::inner_worker_checker(worker, platform_properties)),
+                // Use find to get the most recently used that satisfies the properties.
+                WorkerAllocationStrategy::MostRecentlyUsed => workers_iter
+                    .find(|worker| Self::inner_worker_checker(worker, platform_properties)),
+            };
         workers_iter.map(|(_, w)| w.id.clone())
     }
 

nativelink-util/Cargo.toml

@@ -5,6 +5,9 @@ edition = "2024"
 name = "nativelink-util"
 version = "0.7.0"
 
+[features]
+worker_find_logging = []
+
 [dependencies]
 nativelink-config = { path = "../nativelink-config" }
 nativelink-error = { path = "../nativelink-error" }

nativelink-util/src/platform_properties.rs

@@ -21,6 +21,8 @@ use nativelink_metric::{
 use nativelink_proto::build::bazel::remote::execution::v2::Platform as ProtoPlatform;
 use nativelink_proto::build::bazel::remote::execution::v2::platform::Property as ProtoProperty;
 use serde::{Deserialize, Serialize};
+#[cfg(feature = "worker_find_logging")]
+use tracing::info;
 
 /// `PlatformProperties` helps manage the configuration of platform properties to
 /// keys and types. The scheduler uses these properties to decide what jobs
@@ -47,9 +49,19 @@ impl PlatformProperties {
         for (property, check_value) in &self.properties {
             if let Some(worker_value) = worker_properties.properties.get(property) {
                 if !check_value.is_satisfied_by(worker_value) {
+                    #[cfg(feature = "worker_find_logging")]
+                    {
+                        info!(
+                            "Property mismatch on worker property {property}. {worker_value:?} != {check_value:?}"
+                        );
+                    }
                     return false;
                 }
             } else {
+                #[cfg(feature = "worker_find_logging")]
+                {
+                    info!("Property missing on worker property {property}");
+                }
                 return false;
             }
         }

src/bin/nativelink.rs

@@ -838,6 +838,10 @@ fn get_config() -> Result<CasConfig, Error> {
 }
 
 fn main() -> Result<(), Box<dyn core::error::Error>> {
+    if cfg!(feature = "worker_find_logging") {
+        info!("worker_find_logging enabled");
+    }
+
     let mut cfg = get_config()?;
 
     let global_cfg = if let Some(global_cfg) = &mut cfg.global {