Fixup

Author: dkostyrev

nativelink-scheduler/src/api_worker_scheduler.rs

@@ -15,7 +15,7 @@
 use core::ops::{Deref, DerefMut};
 use core::sync::atomic::{AtomicU64, Ordering};
 use core::time::Duration;
-use std::collections::{HashMap, HashSet};
+use std::collections::HashMap;
 use std::sync::Arc;
 use std::time::{Instant, UNIX_EPOCH};
 

nativelink-scheduler/src/awaited_action_db/mod.rs

@@ -16,7 +16,6 @@ use core::cmp;
 use core::ops::Bound;
 use core::time::Duration;
 use std::collections::HashMap;
-use std::iter::Map;
 use std::sync::Arc;
 
 pub use awaited_action::{AwaitedAction, AwaitedActionSortKey};

nativelink-scheduler/src/memory_awaited_action_db.rs

@@ -16,10 +16,9 @@ use core::ops::{Bound, RangeBounds};
 use core::time::Duration;
 use std::collections::hash_map::Entry;
 use std::collections::{BTreeMap, BTreeSet, HashMap};
-use std::iter::Map;
 use std::sync::Arc;
 
-use async_lock::{Mutex, RwLock};
+use async_lock::RwLock;
 use futures::{FutureExt, Stream};
 use nativelink_config::stores::EvictionPolicy;
 use nativelink_error::{Code, Error, ResultExt, error_if, make_err};

nativelink-scheduler/src/simple_scheduler.rs

@@ -43,13 +43,12 @@ use tokio::time::Duration;
 use tracing::{debug, error, info, info_span, warn};
 
 use crate::api_worker_scheduler::ApiWorkerScheduler;
-use crate::awaited_action_db::{AwaitedAction, AwaitedActionDb, CLIENT_KEEPALIVE_DURATION};
+use crate::awaited_action_db::{AwaitedActionDb, CLIENT_KEEPALIVE_DURATION};
 use crate::platform_property_manager::PlatformPropertyManager;
 use crate::simple_scheduler_state_manager::{SchedulerStateManager, SimpleSchedulerStateManager};
 use crate::worker::{ActionInfoWithProps, ActionsState, Worker, WorkerState, WorkerTimestamp};
 use crate::worker_registry::WorkerRegistry;
 use crate::worker_scheduler::WorkerScheduler;
-use nativelink_util::metrics::StoreType::Metrics;
 use serde::Serialize;
 use nativelink_util::metrics::EXECUTION_METRICS;
 

nativelink-scheduler/src/store_awaited_action_db.rs

@@ -37,7 +37,7 @@ use nativelink_util::store_trait::{
 };
 use nativelink_util::task::JoinHandleDropGuard;
 use tokio::sync::Notify;
-use tracing::{error, info, warn};
+use tracing::{error, warn};
 
 use crate::awaited_action_db::{
     AwaitedAction, AwaitedActionDb, AwaitedActionSubscriber, CLIENT_KEEPALIVE_DURATION,

nativelink-scheduler/src/worker.rs

@@ -27,7 +27,6 @@ use nativelink_util::metrics_utils::{AsyncCounterWrapper, CounterWithTime, FuncC
 use nativelink_util::platform_properties::{PlatformProperties, PlatformPropertyValue};
 use serde::Serialize;
 use tokio::sync::mpsc::UnboundedSender;
-use crate::awaited_action_db::AwaitedAction;
 
 pub type WorkerTimestamp = u64;
 
@@ -61,7 +60,7 @@ pub struct PendingActionInfoData {
     pub action_info: ActionInfoWithProps,
 }
 
-#[derive(Serialize)]
+#[derive(Serialize, Debug)]
 pub struct WorkerState {
     pub id: WorkerId,
     pub platform_properties: PlatformProperties,
@@ -71,7 +70,7 @@ pub struct WorkerState {
     pub is_draining: bool,
 }
 
-#[derive(Serialize)]
+#[derive(Serialize, Debug)]
 pub struct ActionsState {
     pub executing: usize,
     pub queued: usize,

nativelink-store/src/fast_slow_store.rs

@@ -18,7 +18,7 @@ use core::ops::Range;
 use core::pin::Pin;
 use std::collections::HashMap;
 use std::ffi::OsString;
-use std::sync::{Arc, LazyLock, Weak};
+use std::sync::{Arc, Weak};
 
 use async_trait::async_trait;
 use futures::{FutureExt, join};
@@ -34,7 +34,6 @@ use nativelink_util::store_trait::{
     RemoveItemCallback, Store, StoreDriver, StoreKey, StoreLike, StoreOptimizations,
     UploadSizeInfo, slow_update_store_with_file,
 };
-use opentelemetry::{InstrumentationScope, global, metrics};
 use parking_lot::Mutex;
 use tokio::sync::OnceCell;
 use tracing::{debug, trace, warn};

nativelink-store/src/metrics_store.rs

@@ -1,10 +1,7 @@
-use crate::callback_utils::RemoveItemCallbackHolder;
 use crate::filesystem_store::FilesystemStore;
-use crate::memory_store::MemoryStore;
 use async_trait::async_trait;
 use nativelink_error::Error;
 use nativelink_metric::MetricsComponent;
-use nativelink_proto::build_event_stream::File;
 use nativelink_util::buf_channel::{DropCloserReadHalf, DropCloserWriteHalf};
 use nativelink_util::health_utils::{HealthStatus, HealthStatusIndicator};
 use nativelink_util::metrics::{StoreMetricAttrs, StoreType, STORE_METRICS};

nativelink-store/src/redis_store.rs

@@ -21,8 +21,6 @@ use core::str::FromStr;
 use core::time::Duration;
 use std::borrow::Cow;
 use std::collections::HashSet;
-use std::collections::HashMap;
-use std::ops::Index;
 use std::sync::{Arc, Weak};
 use std::time::Instant;
 
@@ -32,22 +30,6 @@ use const_format::formatcp;
 use futures::stream::FuturesUnordered;
 use futures::{Stream, StreamExt, TryFutureExt, TryStreamExt, future};
 use itertools::izip;
-use fred::clients::SubscriberClient;
-use fred::interfaces::{ClientLike, KeysInterface, PubsubInterface};
-use fred::prelude::{Client, EventInterface, HashesInterface, RediSearchInterface};
-use fred::types::config::{
-    Config as RedisConfig, ConnectionConfig, PerformanceConfig, ReconnectPolicy, UnresponsiveConfig,
-};
-use fred::types::redisearch::{
-    AggregateOperation, FtAggregateOptions, FtCreateOptions, IndexKind, Load, ReducerFunc,
-    SearchField, SearchReducer, SearchSchema, SearchSchemaKind, WithCursor,
-};
-use fred::types::scan::Scanner;
-use fred::types::scripts::Script;
-use fred::types::{Builder, Key as RedisKey, Map as RedisMap, SortOrder, Value as RedisValue};
-use futures::stream::FuturesUnordered;
-use futures::{FutureExt, Stream, StreamExt, TryStreamExt, future};
-use itertools::{Itertools, izip};
 use nativelink_config::stores::{RedisMode, RedisSpec};
 use nativelink_error::{Code, Error, ResultExt, make_err, make_input_err};
 use nativelink_metric::MetricsComponent;

nativelink-util/BUILD.bazel

@@ -126,6 +126,7 @@ rust_test_suite(
         "//nativelink-error",
         "//nativelink-metric",
         "//nativelink-proto",
+        "//nativelink-worker",
         "@crates//:axum",
         "@crates//:bytes",
         "@crates//:futures",