Merge branch 'main' into mongo-username-password

Author: palfrey

.github/workflows/native-cargo.yaml

@@ -49,3 +49,7 @@ jobs:
         run: cargo test --all --profile=smol
         env:
           NATIVELINK_TEST_MONGO_URL: ${{ secrets.NATIVELINK_TEST_MONGO_URL}}
+
+      # 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.lock

@@ -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" }

Cargo.toml

@@ -88,7 +88,7 @@
         src = pkgs.lib.cleanSourceWith {
           src = (craneLibFor pkgs).path ./.;
           filter = path: type:
-            (builtins.match "^.*(examples/.+\.json5|data/SekienAkashita\.jpg|nativelink-config/README\.md)" path != null)
+            (builtins.match "^.*(examples/.+\.json5|data/.+|nativelink-config/README\.md)" path != null)
             || ((craneLibFor pkgs).filterCargoSources path type);
         };
 
@@ -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 =

flake.nix

@@ -1,5 +1,5 @@
 {
-    "ABI_LIBC_VERSION": "glibc_2.35",
-    "BAZEL_TARGET_LIBC": "glibc_2.35",
-    "BAZEL_HOST_SYSTEM": "x86_64-unknown-linux-gnu"
+  "ABI_LIBC_VERSION": "glibc_2.35",
+  "BAZEL_HOST_SYSTEM": "x86_64-unknown-linux-gnu",
+  "BAZEL_TARGET_LIBC": "glibc_2.35"
 }

local-remote-execution/cpp_env.json

@@ -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/Cargo.toml

@@ -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-scheduler/src/api_worker_scheduler.rs

@@ -89,6 +89,7 @@ rust_test_suite(
     name = "integration",
     timeout = "short",
     srcs = [
+        "tests/action_messages_test.rs",
         "tests/buf_channel_test.rs",
         "tests/channel_body_for_tests_test.rs",
         "tests/common_test.rs",
@@ -104,6 +105,8 @@ rust_test_suite(
     ],
     compile_data = [
         "tests/data/SekienAkashita.jpg",
+        "tests/data/action_message_cachable_060.json",
+        "tests/data/action_message_uncachable_060.json",
     ],
     proc_macro_deps = [
         "//nativelink-macro",

nativelink-util/BUILD.bazel

@@ -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" }