From cdea7df2b0d71abc143d1d070143f7eadef22985 Mon Sep 17 00:00:00 2001 From: LinZhihao-723 Date: Fri, 7 Aug 2026 16:41:07 -0400 Subject: [PATCH 1/6] Done. --- tools/deployment/spider-helm/values.yaml | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/tools/deployment/spider-helm/values.yaml b/tools/deployment/spider-helm/values.yaml index fe7d60f9..080e6367 100644 --- a/tools/deployment/spider-helm/values.yaml +++ b/tools/deployment/spider-helm/values.yaml @@ -41,8 +41,8 @@ spiderConfig: execution_manager: connection_pool_size: 4 liveness: - scheduler_heartbeat_interval_sec: 10 - storage_heartbeat_interval_sec: 10 + scheduler_heartbeat_interval_sec: 1 + storage_heartbeat_interval_sec: 1 log_level: "INFO" scheduler_poll_wait_ms: 1000 task_executor: @@ -59,12 +59,12 @@ spiderConfig: scheduler: policy: "round_robin" config: - active_job_queue_capacity: 64 + active_job_queue_capacity: 16 cleanup_ready_task_capacity: 256 commit_ready_task_capacity: 256 - dispatch_queue_capacity: 64 + dispatch_queue_capacity: 16 finalizing_job_expiration_timeout_sec: 300 - ready_task_capacity: 65536 + ready_task_capacity: 1048576 storage_poll_timeout_ms: 10 tick_interval_ms: 5 From 1c7977da4b1913cb3eb8bd1753358858bcfa6fa7 Mon Sep 17 00:00:00 2001 From: LinZhihao-723 Date: Fri, 7 Aug 2026 17:05:18 -0400 Subject: [PATCH 2/6] Done. --- components/spider-storage/src/state/runtime.rs | 16 ++++++++-------- .../tests/runtime_recovery_test.rs | 8 ++++---- .../spider-helm/templates/configmap.yaml | 2 +- tools/deployment/spider-helm/values.yaml | 2 +- 4 files changed, 14 insertions(+), 14 deletions(-) diff --git a/components/spider-storage/src/state/runtime.rs b/components/spider-storage/src/state/runtime.rs index bada722d..53ad4c41 100644 --- a/components/spider-storage/src/state/runtime.rs +++ b/components/spider-storage/src/state/runtime.rs @@ -29,13 +29,13 @@ use crate::task_instance_pool::create_task_instance_pool; /// Runtime configuration for the storage service. #[derive(Clone, Debug, Deserialize)] pub struct RuntimeConfig { - pub db_config: DatabaseConfig, + pub db: DatabaseConfig, #[serde(default)] - pub ready_queue_config: ReadyQueueConfig, + pub ready_queue: ReadyQueueConfig, #[serde(default)] - pub task_instance_pool_config: TaskInstancePoolConfig, + pub task_instance_pool: TaskInstancePoolConfig, #[serde(default)] - pub job_cache_gc_config: JobCacheGcConfig, + pub job_cache_gc: JobCacheGcConfig, } /// Runtime state for the storage service. @@ -153,15 +153,15 @@ pub async fn create_runtime( StorageServerError, > { let cancellation_token = CancellationToken::new(); - let db = MariaDbStorageConnector::connect(&config.db_config).await?; + let db = MariaDbStorageConnector::connect(&config.db).await?; let session_id = db.session_id(); let (ready_queue_sender, ready_queue_receiver) = - create_ready_queue(&config.ready_queue_config).map_err(CacheError::from)?; + create_ready_queue(&config.ready_queue).map_err(CacheError::from)?; let (task_instance_pool_connector, task_instance_pool_join_handle) = create_task_instance_pool( ready_queue_sender.clone(), db.clone(), cancellation_token.clone(), - &config.task_instance_pool_config, + &config.task_instance_pool, ) .map_err(CacheError::from)?; @@ -174,7 +174,7 @@ pub async fn create_runtime( let (job_cache_gc_handle, job_cache_gc_join_handle) = create_job_cache_gc( job_cache.clone(), cancellation_token.clone(), - &config.job_cache_gc_config, + &config.job_cache_gc, ) .map_err(CacheError::from)?; let service_state = ServiceState::new(ServiceStateParams { diff --git a/components/spider-storage/tests/runtime_recovery_test.rs b/components/spider-storage/tests/runtime_recovery_test.rs index 012f0cee..870c53ee 100644 --- a/components/spider-storage/tests/runtime_recovery_test.rs +++ b/components/spider-storage/tests/runtime_recovery_test.rs @@ -201,10 +201,10 @@ async fn restarted_storage_cache_recovers_cleanup_ready_job() -> anyhow::Result< /// while other configurations set to default. fn create_runtime_config() -> RuntimeConfig { RuntimeConfig { - db_config: create_mariadb_config(), - ready_queue_config: ReadyQueueConfig::default(), - task_instance_pool_config: TaskInstancePoolConfig::default(), - job_cache_gc_config: JobCacheGcConfig::default(), + db: create_mariadb_config(), + ready_queue: ReadyQueueConfig::default(), + task_instance_pool: TaskInstancePoolConfig::default(), + job_cache_gc: JobCacheGcConfig::default(), } } diff --git a/tools/deployment/spider-helm/templates/configmap.yaml b/tools/deployment/spider-helm/templates/configmap.yaml index 4dfc7502..7b3a3894 100644 --- a/tools/deployment/spider-helm/templates/configmap.yaml +++ b/tools/deployment/spider-helm/templates/configmap.yaml @@ -37,7 +37,7 @@ data: host: "0.0.0.0" port: {{ .Values.spiderConfig.storage.port }} runtime: - db_config: + db: host: {{ include "spider.databaseHost" . | quote }} max_connections: {{ .Values.spiderConfig.database.max_connections }} name: {{ .Values.spiderConfig.database.name | quote }} diff --git a/tools/deployment/spider-helm/values.yaml b/tools/deployment/spider-helm/values.yaml index 080e6367..086df268 100644 --- a/tools/deployment/spider-helm/values.yaml +++ b/tools/deployment/spider-helm/values.yaml @@ -72,7 +72,7 @@ spiderConfig: log_level: "INFO" port: 50051 runtime: - ready_queue_config: + ready_queue: cleanup_capacity: 256 commit_capacity: 256 task_capacity: 1048576 From ab46c64d67ba4ce2f83b693301ca912e639af97d Mon Sep 17 00:00:00 2001 From: LinZhihao-723 Date: Fri, 7 Aug 2026 17:05:51 -0400 Subject: [PATCH 3/6] Update chart version. --- tools/deployment/spider-helm/Chart.yaml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tools/deployment/spider-helm/Chart.yaml b/tools/deployment/spider-helm/Chart.yaml index 8de00ec2..5b15c1c8 100644 --- a/tools/deployment/spider-helm/Chart.yaml +++ b/tools/deployment/spider-helm/Chart.yaml @@ -2,7 +2,7 @@ apiVersion: "v2" name: "spider" description: "A Helm chart for the Spider Huntsman deployment" type: "application" -version: "0.1.8" +version: "0.1.9" appVersion: "0.1.0-dev" home: "https://github.com/y-scope/spider" sources: ["https://github.com/y-scope/spider"] From d6ffc054a822f34b9434e1f0bbfdfbfdcc6cb085 Mon Sep 17 00:00:00 2001 From: LinZhihao-723 Date: Fri, 7 Aug 2026 17:06:25 -0400 Subject: [PATCH 4/6] Update chart version. --- tools/deployment/spider-helm/Chart.yaml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tools/deployment/spider-helm/Chart.yaml b/tools/deployment/spider-helm/Chart.yaml index 5b15c1c8..4fa15b4e 100644 --- a/tools/deployment/spider-helm/Chart.yaml +++ b/tools/deployment/spider-helm/Chart.yaml @@ -2,7 +2,7 @@ apiVersion: "v2" name: "spider" description: "A Helm chart for the Spider Huntsman deployment" type: "application" -version: "0.1.9" +version: "0.1.10" appVersion: "0.1.0-dev" home: "https://github.com/y-scope/spider" sources: ["https://github.com/y-scope/spider"] From ac0edca0a23745529689db612fa20ec61392d8e3 Mon Sep 17 00:00:00 2001 From: LinZhihao-723 Date: Fri, 7 Aug 2026 19:29:40 -0400 Subject: [PATCH 5/6] Done. --- tools/deployment/spider-helm/Chart.yaml | 2 +- tools/deployment/spider-helm/templates/configmap.yaml | 7 +++---- tools/deployment/spider-helm/values.yaml | 10 ++++++++++ 3 files changed, 14 insertions(+), 5 deletions(-) diff --git a/tools/deployment/spider-helm/Chart.yaml b/tools/deployment/spider-helm/Chart.yaml index 4fa15b4e..46de908b 100644 --- a/tools/deployment/spider-helm/Chart.yaml +++ b/tools/deployment/spider-helm/Chart.yaml @@ -2,7 +2,7 @@ apiVersion: "v2" name: "spider" description: "A Helm chart for the Spider Huntsman deployment" type: "application" -version: "0.1.10" +version: "0.1.11" appVersion: "0.1.0-dev" home: "https://github.com/y-scope/spider" sources: ["https://github.com/y-scope/spider"] diff --git a/tools/deployment/spider-helm/templates/configmap.yaml b/tools/deployment/spider-helm/templates/configmap.yaml index 7b3a3894..e5707305 100644 --- a/tools/deployment/spider-helm/templates/configmap.yaml +++ b/tools/deployment/spider-helm/templates/configmap.yaml @@ -26,10 +26,9 @@ data: advertised_endpoint: host: {{ include "spider.componentFullname" (dict "root" . "component" "scheduler") | quote }} port: {{ .Values.spiderConfig.scheduler.port }} - scheduler: - policy: {{ .Values.spiderConfig.scheduler.runtime.scheduler.policy | quote }} - config: - {{- toYaml .Values.spiderConfig.scheduler.runtime.scheduler.config | nindent 10 }} + {{- with .Values.spiderConfig.scheduler.runtime }} + {{- toYaml . | nindent 6 }} + {{- end }} storage_endpoint: host: {{ include "spider.componentFullname" (dict "root" . "component" "storage") | quote }} port: {{ .Values.spiderConfig.storage.port }} diff --git a/tools/deployment/spider-helm/values.yaml b/tools/deployment/spider-helm/values.yaml index 086df268..41d52609 100644 --- a/tools/deployment/spider-helm/values.yaml +++ b/tools/deployment/spider-helm/values.yaml @@ -56,6 +56,9 @@ spiderConfig: log_level: "INFO" port: 50052 runtime: + em_registry: + dead_em_cutoff_sec: 30 + liveness_tracking_interval_ms: 1000 scheduler: policy: "round_robin" config: @@ -72,10 +75,17 @@ spiderConfig: log_level: "INFO" port: 50051 runtime: + job_cache_gc: + gc_interval_sec: 30 + terminated_job_retention_sec: 300 ready_queue: cleanup_capacity: 256 commit_capacity: 256 task_capacity: 1048576 + task_instance_pool: + execution_manager_stale_cutoff_sec: 60 + gc_interval_sec: 30 + message_channel_capacity: 128 worker: extra_envs: [] From a9ad42f08e140ee3eee4802a783ebd8e0cc7c213 Mon Sep 17 00:00:00 2001 From: LinZhihao-723 Date: Mon, 10 Aug 2026 15:03:51 -0400 Subject: [PATCH 6/6] Done. --- components/spider-scheduler/src/lib.rs | 4 +- components/spider-storage/src/cache/error.rs | 8 +- components/spider-storage/src/cache/job.rs | 127 +++++++------ components/spider-storage/src/grpc.rs | 54 +++--- .../src/{ready_queue.rs => inbound_queue.rs} | 114 ++++++------ components/spider-storage/src/lib.rs | 2 +- .../spider-storage/src/state/job_cache.rs | 104 +++++++---- .../spider-storage/src/state/job_cache_gc.rs | 22 +-- .../spider-storage/src/state/runtime.rs | 60 +++--- .../spider-storage/src/state/service.rs | 171 +++++++++--------- .../spider-storage/src/state/test_utils.rs | 8 +- .../spider-storage/src/task_instance_pool.rs | 122 +++++++------ .../tests/runtime_recovery_test.rs | 70 +++---- .../spider-storage/tests/scheduling_infra.rs | 58 +++--- tools/deployment/spider-helm/Chart.yaml | 2 +- tools/deployment/spider-helm/values.yaml | 8 +- 16 files changed, 490 insertions(+), 444 deletions(-) rename components/spider-storage/src/{ready_queue.rs => inbound_queue.rs} (76%) diff --git a/components/spider-scheduler/src/lib.rs b/components/spider-scheduler/src/lib.rs index d7aac238..a39e4d7b 100644 --- a/components/spider-scheduler/src/lib.rs +++ b/components/spider-scheduler/src/lib.rs @@ -6,11 +6,11 @@ //! order* and *with what throttling* ready tasks are offered to the fleet. //! //! The crate defines three trait seams wired into a single pipeline — a storage client that polls -//! the ready queue, a core that makes serial decisions, and a dispatching queue that fans those +//! the inbound queue, a core that makes serial decisions, and a dispatching queue that fans those //! decisions out to execution managers: //! //! ```text -//! storage ── authoritative ready queue (owned by the storage layer, not this crate) +//! storage ── authoritative inbound queue (owned by the storage layer, not this crate) //! │ //! │ poll_ready / poll_commit_ready / poll_cleanup_ready (SchedulerStorageClient) //! ▼ diff --git a/components/spider-storage/src/cache/error.rs b/components/spider-storage/src/cache/error.rs index 48236403..517b50d3 100644 --- a/components/spider-storage/src/cache/error.rs +++ b/components/spider-storage/src/cache/error.rs @@ -79,8 +79,8 @@ pub enum InternalError { #[error("task instance pool corrupted: {0}")] TaskInstancePoolCorrupted(String), - #[error("invalid ready-queue config: {0}")] - ReadyQueueInvalidConfig(&'static str), + #[error("invalid inbound-queue config: {0}")] + InboundQueueInvalidConfig(&'static str), #[error("invalid task instance pool config: {0}")] TaskInstancePoolInvalidConfig(&'static str), @@ -88,8 +88,8 @@ pub enum InternalError { #[error("invalid job cache GC config: {0}")] JobCacheGcInvalidConfig(&'static str), - #[error("ready queue channel is closed")] - ReadyQueueChannelClosed, + #[error("inbound queue channel is closed")] + InboundQueueChannelClosed, #[error("invalid recoverable job context: {0}")] InvalidRecoverableJobContext(String), diff --git a/components/spider-storage/src/cache/job.rs b/components/spider-storage/src/cache/job.rs index 1b90c687..84ccec34 100644 --- a/components/spider-storage/src/cache/job.rs +++ b/components/spider-storage/src/cache/job.rs @@ -24,36 +24,38 @@ use crate::cache::error::StaleStateError; use crate::cache::task::TaskGraph; use crate::db::InternalJobOrchestration; use crate::db::RecoverableJobContext; +use crate::inbound_queue::InboundQueueSender; use crate::job_submission::ValidatedJobSubmission; -use crate::ready_queue::ReadyQueueSender; use crate::task_instance_pool::TaskInstanceMetadata; use crate::task_instance_pool::TaskInstancePoolConnector; /// A shareable control block for a job. /// -/// All mutable state, including the task graph, connectors, and queue sender, is held inside the +/// All mutable state, including the task graph, connectors, and inbound-queue sender, is held +/// inside the /// underlying [`JobExecutionState`] and protected by [`JobExecutionStateHandle`]'s read-write lock. /// /// # Type Parameters /// -/// * `ReadyQueueSenderType` - The type of the ready queue sender. +/// * `InboundQueueSenderType` - The type of the inbound queue sender. /// * `DbConnectorType` - The type of the DB-layer connector. /// * `TaskInstancePoolConnectorType` - The type of the task instance pool connector. #[derive(Clone)] pub struct SharedJobControlBlock< - ReadyQueueSenderType: ReadyQueueSender, + InboundQueueSenderType: InboundQueueSender, DbConnectorType: InternalJobOrchestration, TaskInstancePoolConnectorType: TaskInstancePoolConnector, > { - inner: - Arc>, + inner: Arc< + JobControlBlock, + >, } impl< - ReadyQueueSenderType: ReadyQueueSender, + InboundQueueSenderType: InboundQueueSender, DbConnectorType: InternalJobOrchestration, TaskInstancePoolConnectorType: TaskInstancePoolConnector, -> SharedJobControlBlock +> SharedJobControlBlock { /// Factory function. /// @@ -70,7 +72,7 @@ impl< id: JobId, owner_id: ResourceGroupId, job_submission: ValidatedJobSubmission, - ready_queue_sender: ReadyQueueSenderType, + inbound_queue_sender: InboundQueueSenderType, db_connector: DbConnectorType, task_instance_pool_connector: TaskInstancePoolConnectorType, ) -> Result { @@ -80,7 +82,7 @@ impl< state: JobState::Ready, task_graph, num_incomplete_tasks: AtomicUsize::new(num_tasks), - ready_queue_sender, + inbound_queue_sender, db_connector, task_instance_pool_connector, }; @@ -101,7 +103,7 @@ impl< /// # NOTE /// /// * This constructor does not mutate the storage states. - /// * This constructor does not send recovered tasks to the ready-queue. + /// * This constructor does not send recovered tasks to the inbound-queue. /// /// # Returns /// @@ -119,7 +121,7 @@ impl< /// * Forwards [`TaskGraph::create`]'s return values on failure. pub async fn recover( recoverable_job_context: RecoverableJobContext, - ready_queue_sender: ReadyQueueSenderType, + inbound_queue_sender: InboundQueueSenderType, db_connector: DbConnectorType, task_instance_pool_connector: TaskInstancePoolConnectorType, ) -> Result { @@ -181,7 +183,7 @@ impl< state, task_graph, num_incomplete_tasks: AtomicUsize::new(num_incomplete_tasks), - ready_queue_sender, + inbound_queue_sender, db_connector, task_instance_pool_connector, }; @@ -231,7 +233,7 @@ impl< /// Starts the job. /// - /// Any tasks in [`TaskState::Ready`] will be enqueued to the ready queue on success. + /// Any tasks in [`TaskState::Ready`] will be enqueued to the inbound queue on success. /// /// # Errors /// @@ -239,7 +241,7 @@ impl< /// /// * Forwards [`JobExecutionStateHandle::write_ready`]'s return values on failure. /// * Forwards [`InternalJobOrchestration::start`]'s return values on failure. - /// * Forwards [`ReadyQueueSender::send_task_ready`]'s return values on failure. + /// * Forwards [`InboundQueueSender::send_task_ready`]'s return values on failure. pub async fn start(&self) -> Result<(), CacheError> { let jcb = &self.inner; let mut job = jcb.job_execution_state.write_ready().await?; @@ -257,22 +259,22 @@ impl< // JCB. If it happens to travel fast enough to go into the scheduler and then the executor, // the request from the executor for registering task instances will be blocked until this // method returns. - job.ready_queue_sender + job.inbound_queue_sender .send_task_ready(jcb.owner_id, jcb.id, ready_task_indices) .await?; drop(job); Ok(()) } - /// Resends all ready tasks to the ready queue. + /// Resends all ready tasks to the inbound queue. /// /// The method handles the following job states: /// /// * [`JobState::Running`] — all tasks in [`TaskState::Ready`] via - /// [`ReadyQueueSender::send_task_ready`]. - /// * [`JobState::CommitReady`] — the commit task via [`ReadyQueueSender::send_commit_ready`]. + /// [`InboundQueueSender::send_task_ready`]. + /// * [`JobState::CommitReady`] — the commit task via [`InboundQueueSender::send_commit_ready`]. /// * [`JobState::CleanupReady`] — the cleanup task via - /// [`ReadyQueueSender::send_cleanup_ready`]. + /// [`InboundQueueSender::send_cleanup_ready`]. /// /// For other job states, this method is a no-op and returns `Ok(())`. /// @@ -280,16 +282,16 @@ impl< /// /// Returns an error if: /// - /// * Forwards [`ReadyQueueSender::send_task_ready`]'s return values on failure. - /// * Forwards [`ReadyQueueSender::send_commit_ready`]'s return values on failure. - /// * Forwards [`ReadyQueueSender::send_cleanup_ready`]'s return values on failure. + /// * Forwards [`InboundQueueSender::send_task_ready`]'s return values on failure. + /// * Forwards [`InboundQueueSender::send_commit_ready`]'s return values on failure. + /// * Forwards [`InboundQueueSender::send_cleanup_ready`]'s return values on failure. pub async fn resend_ready_tasks(&self) -> Result<(), CacheError> { let jcb = &self.inner; if let Ok(job) = jcb.job_execution_state.read_running().await { let ready_task_indices = job.task_graph.get_all_ready_task_indices().await; if !ready_task_indices.is_empty() { - job.ready_queue_sender + job.inbound_queue_sender .send_task_ready(jcb.owner_id, jcb.id, ready_task_indices) .await?; } @@ -297,14 +299,14 @@ impl< } if let Ok(job) = jcb.job_execution_state.read_commit_ready().await { - job.ready_queue_sender + job.inbound_queue_sender .send_commit_ready(jcb.owner_id, jcb.id) .await?; return Ok(()); } if let Ok(job) = jcb.job_execution_state.read_cleanup_ready().await { - job.ready_queue_sender + job.inbound_queue_sender .send_cleanup_ready(jcb.owner_id, jcb.id) .await?; return Ok(()); @@ -344,8 +346,8 @@ impl< /// Marks the task instance as succeeded. /// /// If all tasks have succeeded, commits the job outputs, transitions the job state, and - /// enqueues the commit task (if any) to the ready queue. Otherwise, if the completed task - /// unblocks any child tasks, those child tasks are enqueued to the ready queue. + /// enqueues the commit task (if any) to the inbound queue. Otherwise, if the completed task + /// unblocks any child tasks, those child tasks are enqueued to the inbound queue. /// /// # Returns /// @@ -365,8 +367,8 @@ impl< /// * Forwards [`JobExecutionStateHandle::read_running`]'s return values on failure. /// * Forwards [`JobExecutionStateHandle::write_running`]'s return values on failure. /// * Forwards [`SharedTaskControlBlock::succeed_task_instance`]'s return values on failure. - /// * Forwards [`ReadyQueueSender::send_task_ready`]'s return values on failure. - /// * Forwards [`ReadyQueueSender::send_commit_ready`]'s return values on failure. + /// * Forwards [`InboundQueueSender::send_task_ready`]'s return values on failure. + /// * Forwards [`InboundQueueSender::send_commit_ready`]'s return values on failure. /// * Forwards [`SharedJobControlBlock::commit_outputs`]'s return values on failure. /// * Forwards [`TaskGraph::read_output_payloads`]'s return values on failure. /// * Forwards [`InternalJobOrchestration::commit_outputs`]'s return values on failure. @@ -396,7 +398,7 @@ impl< ) .into()); } - job.ready_queue_sender + job.inbound_queue_sender .send_task_ready(jcb.owner_id, jcb.id, ready_task_indices) .await?; return Ok(job.state); @@ -420,7 +422,7 @@ impl< JobState::Succeeded }; if has_commit_task { - job.ready_queue_sender + job.inbound_queue_sender .send_commit_ready(jcb.owner_id, jcb.id) .await?; } @@ -500,8 +502,8 @@ impl< /// Marks a task instance as failed. /// - /// If the task has remaining retries, it is re-enqueued to the ready queue. Otherwise, the job - /// transitions to [`JobState::Failed`]. + /// If the task has remaining retries, it is re-enqueued to the inbound queue. Otherwise, the + /// job transitions to [`JobState::Failed`]. /// /// # Returns /// @@ -525,9 +527,9 @@ impl< /// * Forwards [`SharedTaskControlBlock::fail_task_instance`]'s return values on failure. /// * Forwards [`SharedTerminationTaskControlBlock::fail_task_instance`]'s return values on /// failure. - /// * Forwards [`ReadyQueueSender::send_task_ready`]'s return values on failure. - /// * Forwards [`ReadyQueueSender::send_commit_ready`]'s return values on failure. - /// * Forwards [`ReadyQueueSender::send_cleanup_ready`]'s return values on failure. + /// * Forwards [`InboundQueueSender::send_task_ready`]'s return values on failure. + /// * Forwards [`InboundQueueSender::send_commit_ready`]'s return values on failure. + /// * Forwards [`InboundQueueSender::send_cleanup_ready`]'s return values on failure. /// * Forwards [`InternalJobOrchestration::fail`]'s return values on failure. pub async fn fail_task_instance( &self, @@ -546,7 +548,7 @@ impl< .fail_task_instance(task_instance_id, error_message.clone()) .await?; if matches!(task_state, TaskState::Ready | TaskState::Running) { - job.ready_queue_sender + job.inbound_queue_sender .send_task_ready(jcb.owner_id, jcb.id, vec![task_index]) .await?; return Ok(job.state); @@ -561,7 +563,7 @@ impl< .fail_task_instance(task_instance_id, error_message.clone()) .await?; if matches!(task_state, TaskState::Ready | TaskState::Running) { - job.ready_queue_sender + job.inbound_queue_sender .send_commit_ready(jcb.owner_id, jcb.id) .await?; return Ok(job.state); @@ -576,7 +578,7 @@ impl< .fail_task_instance(task_instance_id, error_message.clone()) .await?; if matches!(task_state, TaskState::Ready | TaskState::Running) { - job.ready_queue_sender + job.inbound_queue_sender .send_cleanup_ready(jcb.owner_id, jcb.id) .await?; return Ok(job.state); @@ -618,7 +620,7 @@ impl< /// /// * Forwards [`JobExecutionStateHandle::write_cancellable`]'s return values on failure. /// * Forwards [`InternalJobOrchestration::cancel`]'s return values on failure. - /// * Forwards [`ReadyQueueSender::send_cleanup_ready`]'s return values on failure. + /// * Forwards [`InboundQueueSender::send_cleanup_ready`]'s return values on failure. pub async fn cancel(&self) -> Result { let jcb = &self.inner; let mut job = jcb.job_execution_state.write_cancellable().await?; @@ -632,7 +634,7 @@ impl< job.task_graph.cancel_non_terminal().await; if has_cleanup_task { - job.ready_queue_sender + job.inbound_queue_sender .send_cleanup_ready(jcb.owner_id, jcb.id) .await?; } @@ -656,7 +658,11 @@ impl< /// * Forwards [`SharedTaskControlBlock::register_task_instance`]'s return values on failure. /// * Forwards [`TaskInstancePoolConnector::register_task_instance`]'s return values on failure. async fn create_regular_task_instance( - jcb: &JobControlBlock, + jcb: &JobControlBlock< + InboundQueueSenderType, + DbConnectorType, + TaskInstancePoolConnectorType, + >, task_index: TaskIndex, execution_manager_id: ExecutionManagerId, ) -> Result { @@ -708,7 +714,11 @@ impl< /// * Forwards [`TaskGraph::read_output_payloads`]'s return values on failure. /// * Forwards [`SerializedTaskOutputs::serialize_with_size_hint`]'s return values on failure. async fn create_commit_task_instance( - jcb: &JobControlBlock, + jcb: &JobControlBlock< + InboundQueueSenderType, + DbConnectorType, + TaskInstancePoolConnectorType, + >, execution_manager_id: ExecutionManagerId, ) -> Result { let job = jcb.job_execution_state.read_commit_ready().await?; @@ -766,7 +776,11 @@ impl< /// * Forwards [`TaskInstancePoolConnector::register_termination_task_instance`]'s return values /// on failure. async fn create_cleanup_task_instance( - jcb: &JobControlBlock, + jcb: &JobControlBlock< + InboundQueueSenderType, + DbConnectorType, + TaskInstancePoolConnectorType, + >, execution_manager_id: ExecutionManagerId, ) -> Result { let job = jcb.job_execution_state.read_cleanup_ready().await?; @@ -810,18 +824,18 @@ impl< /// /// # Type Parameters /// -/// * `ReadyQueueSenderType` - The type of the ready queue sender. +/// * `InboundQueueSenderType` - The type of the inbound queue sender. /// * `DbConnectorType` - The type of the DB-layer connector. /// * `TaskInstancePoolConnectorType` - The type of the task instance pool connector. struct JobControlBlock< - ReadyQueueSenderType: ReadyQueueSender, + InboundQueueSenderType: InboundQueueSender, DbConnectorType: InternalJobOrchestration, TaskInstancePoolConnectorType: TaskInstancePoolConnector, > { id: JobId, owner_id: ResourceGroupId, job_execution_state: JobExecutionStateHandle< - ReadyQueueSenderType, + InboundQueueSenderType, DbConnectorType, TaskInstancePoolConnectorType, >, @@ -842,14 +856,14 @@ struct JobControlBlock< /// * Avoid formatting issues, as `rustfmt` does not handle line wrapping well when using more /// descriptive type parameter names in this particular struct. struct JobExecutionStateHandle< - R: ReadyQueueSender, + R: InboundQueueSender, D: InternalJobOrchestration, T: TaskInstancePoolConnector, > { inner: tokio::sync::RwLock>, } -impl +impl JobExecutionStateHandle { /// # Returns @@ -1058,32 +1072,33 @@ impl { state: JobState, task_graph: TaskGraph, num_incomplete_tasks: AtomicUsize, - ready_queue_sender: ReadyQueueSenderType, + inbound_queue_sender: InboundQueueSenderType, db_connector: DbConnectorType, task_instance_pool_connector: TaskInstancePoolConnectorType, } impl< - ReadyQueueSenderType: ReadyQueueSender, + InboundQueueSenderType: InboundQueueSender, DbConnectorType: InternalJobOrchestration, TaskInstancePoolConnectorType: TaskInstancePoolConnector, -> JobExecutionState +> JobExecutionState { /// Ensures that the job is currently in the [`JobState::Running`] state. /// diff --git a/components/spider-storage/src/grpc.rs b/components/spider-storage/src/grpc.rs index e5ca70b8..393fb794 100644 --- a/components/spider-storage/src/grpc.rs +++ b/components/spider-storage/src/grpc.rs @@ -22,8 +22,8 @@ use tonic::Status; use crate::cache::error::CacheError; use crate::db::DbError; use crate::db::DbStorage; -use crate::ready_queue::ReadyQueueEntry; -use crate::ready_queue::ReadyQueueSender; +use crate::inbound_queue::InboundQueueEntry; +use crate::inbound_queue::InboundQueueSender; use crate::state::ServiceState; use crate::state::StorageServerError; use crate::task_instance_pool::TaskInstancePoolConnector; @@ -32,24 +32,24 @@ use crate::task_instance_pool::TaskInstancePoolConnector; /// /// # Type Parameters /// -/// * `ReadyQueueSenderType` - The ready queue sender type. +/// * `InboundQueueSenderType` - The inbound queue sender type. /// * `DbConnectorType` - The database connector type. /// * `TaskInstancePoolConnectorType` - The task instance pool connector type. #[derive(Clone)] pub struct GrpcServiceState< - ReadyQueueSenderType: ReadyQueueSender + 'static, + InboundQueueSenderType: InboundQueueSender + 'static, DbConnectorType: DbStorage + 'static, TaskInstancePoolConnectorType: TaskInstancePoolConnector + 'static, > { - inner: ServiceState, + inner: ServiceState, cancellation_token: CancellationToken, } impl< - ReadyQueueSenderType: ReadyQueueSender + 'static, + InboundQueueSenderType: InboundQueueSender + 'static, DbConnectorType: DbStorage + 'static, TaskInstancePoolConnectorType: TaskInstancePoolConnector + 'static, -> GrpcServiceState +> GrpcServiceState { /// Factory function. /// @@ -58,7 +58,7 @@ impl< /// A new [`GrpcServiceState`] wrapping [`ServiceState`]. #[must_use] pub const fn new( - inner: ServiceState, + inner: ServiceState, cancellation_token: CancellationToken, ) -> Self { Self { @@ -420,21 +420,21 @@ impl< Status::internal("storage service internal error") } - /// Builds a [`storage::ReadyTasks`] message from a batch of ready-queue entries. + /// Builds a [`storage::ReadyTasks`] message from a batch of inbound-queue entries. /// /// # Type Parameters /// - /// * `TaskKindType` - The kind of ready-queue task carried by each entry: + /// * `TaskKindType` - The kind of inbound-queue task carried by each entry: /// * [`spider_core::task::TaskIndex`] for the regular lane. - /// * [`crate::ready_queue::CommitTaskMarker`] for the commit lane. - /// * [`crate::ready_queue::CleanupTaskMarker`] for the cleanup lane. + /// * [`crate::inbound_queue::CommitTaskMarker`] for the commit lane. + /// * [`crate::inbound_queue::CleanupTaskMarker`] for the cleanup lane. /// /// # Returns /// /// A [`storage::ReadyTasks`] carrying the storage session and the flattened ready tasks. fn build_ready_tasks( &self, - entries: Vec>, + entries: Vec>, to_task_id: impl Fn(TaskKindType) -> common::TaskId, ) -> storage::ReadyTasks { let tasks = entries @@ -463,11 +463,11 @@ impl< /// [`GrpcServiceState::job_orchestration_service_error_handler`]. #[async_trait] impl< - ReadyQueueSenderType: ReadyQueueSender + 'static, + InboundQueueSenderType: InboundQueueSender + 'static, DbConnectorType: DbStorage + 'static, TaskInstancePoolConnectorType: TaskInstancePoolConnector + 'static, > JobOrchestrationService - for GrpcServiceState + for GrpcServiceState { async fn register_job( &self, @@ -569,11 +569,11 @@ impl< #[async_trait] impl< - ReadyQueueSenderType: ReadyQueueSender + 'static, + InboundQueueSenderType: InboundQueueSender + 'static, DbConnectorType: DbStorage + 'static, TaskInstancePoolConnectorType: TaskInstancePoolConnector + 'static, > TaskInstanceManagementService - for GrpcServiceState + for GrpcServiceState { async fn register_task_instance( &self, @@ -683,11 +683,11 @@ impl< #[async_trait] impl< - ReadyQueueSenderType: ReadyQueueSender + 'static, + InboundQueueSenderType: InboundQueueSender + 'static, DbConnectorType: DbStorage + 'static, TaskInstancePoolConnectorType: TaskInstancePoolConnector + 'static, > InboundQueueService - for GrpcServiceState + for GrpcServiceState { async fn poll_ready_tasks( &self, @@ -746,11 +746,11 @@ impl< #[async_trait] impl< - ReadyQueueSenderType: ReadyQueueSender + 'static, + InboundQueueSenderType: InboundQueueSender + 'static, DbConnectorType: DbStorage + 'static, TaskInstancePoolConnectorType: TaskInstancePoolConnector + 'static, > ResourceGroupManagementService - for GrpcServiceState + for GrpcServiceState { async fn add_resource_group( &self, @@ -791,11 +791,11 @@ impl< #[async_trait] impl< - ReadyQueueSenderType: ReadyQueueSender + 'static, + InboundQueueSenderType: InboundQueueSender + 'static, DbConnectorType: DbStorage + 'static, TaskInstancePoolConnectorType: TaskInstancePoolConnector + 'static, > ExecutionManagerLivenessService - for GrpcServiceState + for GrpcServiceState { async fn register_execution_manager( &self, @@ -849,11 +849,11 @@ impl< #[async_trait] impl< - ReadyQueueSenderType: ReadyQueueSender + 'static, + InboundQueueSenderType: InboundQueueSender + 'static, DbConnectorType: DbStorage + 'static, TaskInstancePoolConnectorType: TaskInstancePoolConnector + 'static, > SchedulerRegistrationService - for GrpcServiceState + for GrpcServiceState { async fn register_scheduler( &self, @@ -886,11 +886,11 @@ impl< #[async_trait] impl< - ReadyQueueSenderType: ReadyQueueSender + 'static, + InboundQueueSenderType: InboundQueueSender + 'static, DbConnectorType: DbStorage + 'static, TaskInstancePoolConnectorType: TaskInstancePoolConnector + 'static, > SessionManagementService - for GrpcServiceState + for GrpcServiceState { async fn get_session( &self, diff --git a/components/spider-storage/src/ready_queue.rs b/components/spider-storage/src/inbound_queue.rs similarity index 76% rename from components/spider-storage/src/ready_queue.rs rename to components/spider-storage/src/inbound_queue.rs index 14fc3f61..e85e6cdb 100644 --- a/components/spider-storage/src/ready_queue.rs +++ b/components/spider-storage/src/inbound_queue.rs @@ -1,13 +1,13 @@ -//! In-memory ready queue for schedulable tasks. +//! In-memory inbound queue for schedulable tasks. //! //! The queue is a set of three independent MPMC async channels — one for regular tasks, one for -//! commit tasks, and one for cleanup tasks. Each channel carries a [`ReadyQueueEntry`] +//! commit tasks, and one for cleanup tasks. Each channel carries an [`InboundQueueEntry`] //! parameterized by the lane-specific task kind: [`TaskIndex`] for the regular lane, //! [`CommitTaskMarker`] for the commit lane, and [`CleanupTaskMarker`] for the cleanup lane. //! -//! [`ReadyQueueSender`] routes each send to the matching channel, and [`ReadyQueueReceiverHandle`] -//! exposes three `recv_*` methods that each read from one channel with a -//! `(max_items, wait_duration)` signature. +//! [`InboundQueueSender`] routes each send to the matching channel, and +//! [`InboundQueueReceiverHandle`] exposes three `recv_*` methods that each read from one channel +//! with a `(max_items, wait_duration)` signature. use std::time::Duration; @@ -21,15 +21,15 @@ use spider_core::types::id::ResourceGroupId; use crate::cache::error::InternalError; -/// Marker type for commit-task ready entries. +/// Marker type for commit-task inbound-queue entries. #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] pub struct CommitTaskMarker; -/// Marker type for cleanup-task ready entries. +/// Marker type for cleanup-task inbound-queue entries. #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] pub struct CleanupTaskMarker; -/// A ready queue entry. +/// An inbound queue entry. /// /// # Type Parameters /// @@ -38,7 +38,7 @@ pub struct CleanupTaskMarker; /// * [`CommitTaskMarker`] for commit tasks, and /// * [`CleanupTaskMarker`] for cleanup tasks. #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] -pub struct ReadyQueueEntry { +pub struct InboundQueueEntry { /// The owning resource group for the ready job. pub resource_group_id: ResourceGroupId, /// The job that became schedulable. @@ -47,10 +47,10 @@ pub struct ReadyQueueEntry { pub task_kind: TaskKind, } -/// Configuration of a ready queue. +/// Configuration of an inbound queue. #[derive(Debug, Clone, Copy, Deserialize)] #[serde(default)] -pub struct ReadyQueueConfig { +pub struct InboundQueueConfig { /// The capacity of the task lane. Must be greater than zero. pub task_capacity: usize, /// The capacity of the commit lane. Must be greater than zero. @@ -59,7 +59,7 @@ pub struct ReadyQueueConfig { pub cleanup_capacity: usize, } -impl Default for ReadyQueueConfig { +impl Default for InboundQueueConfig { fn default() -> Self { Self { task_capacity: DEFAULT_TASK_CAPACITY, @@ -69,28 +69,28 @@ impl Default for ReadyQueueConfig { } } -impl ReadyQueueConfig { +impl InboundQueueConfig { /// Validates the config. /// /// # Errors /// /// Returns an error if: /// - /// * [`InternalError::ReadyQueueInvalidConfig`] if any of the configured capacity is 0. + /// * [`InternalError::InboundQueueInvalidConfig`] if any of the configured capacity is 0. const fn validate(self) -> Result<(), InternalError> { const ERROR_MESSAGE: &str = "capacity must be greater than 0"; if self.task_capacity == 0 || self.commit_capacity == 0 || self.cleanup_capacity == 0 { - return Err(InternalError::ReadyQueueInvalidConfig(ERROR_MESSAGE)); + return Err(InternalError::InboundQueueInvalidConfig(ERROR_MESSAGE)); } Ok(()) } } -/// Connector for publishing task execution events to the ready queue. +/// Connector for publishing task execution events to the inbound queue. /// /// This trait is invoked by the cache layer to enqueue tasks that are ready for scheduling. #[async_trait] -pub trait ReadyQueueSender: Clone + Send + Sync { +pub trait InboundQueueSender: Clone + Send + Sync { /// Enqueues a batch of tasks for the specified job which are ready to be scheduled. Each task /// index becomes one entry on the task lane. /// @@ -145,16 +145,16 @@ pub trait ReadyQueueSender: Clone + Send + Sync { ) -> Result<(), InternalError>; } -/// A shareable ready-queue sender backed by three MPMC channels. +/// A shareable inbound-queue sender backed by three MPMC channels. #[derive(Clone)] -pub struct ReadyQueueSenderHandle { - task: Sender>, - commit: Sender>, - cleanup: Sender>, +pub struct InboundQueueSenderHandle { + task: Sender>, + commit: Sender>, + cleanup: Sender>, } #[async_trait] -impl ReadyQueueSender for ReadyQueueSenderHandle { +impl InboundQueueSender for InboundQueueSenderHandle { async fn send_task_ready( &self, resource_group_id: ResourceGroupId, @@ -162,7 +162,7 @@ impl ReadyQueueSender for ReadyQueueSenderHandle { task_indices: Vec, ) -> Result<(), InternalError> { for task_index in task_indices { - let entry = ReadyQueueEntry { + let entry = InboundQueueEntry { resource_group_id, job_id, task_kind: task_index, @@ -170,7 +170,7 @@ impl ReadyQueueSender for ReadyQueueSenderHandle { self.task .send(entry) .await - .map_err(|_| InternalError::ReadyQueueChannelClosed)?; + .map_err(|_| InternalError::InboundQueueChannelClosed)?; } Ok(()) } @@ -180,7 +180,7 @@ impl ReadyQueueSender for ReadyQueueSenderHandle { resource_group_id: ResourceGroupId, job_id: JobId, ) -> Result<(), InternalError> { - let entry = ReadyQueueEntry { + let entry = InboundQueueEntry { resource_group_id, job_id, task_kind: CommitTaskMarker, @@ -188,7 +188,7 @@ impl ReadyQueueSender for ReadyQueueSenderHandle { self.commit .send(entry) .await - .map_err(|_| InternalError::ReadyQueueChannelClosed) + .map_err(|_| InternalError::InboundQueueChannelClosed) } async fn send_cleanup_ready( @@ -196,7 +196,7 @@ impl ReadyQueueSender for ReadyQueueSenderHandle { resource_group_id: ResourceGroupId, job_id: JobId, ) -> Result<(), InternalError> { - let entry = ReadyQueueEntry { + let entry = InboundQueueEntry { resource_group_id, job_id, task_kind: CleanupTaskMarker, @@ -204,28 +204,28 @@ impl ReadyQueueSender for ReadyQueueSenderHandle { self.cleanup .send(entry) .await - .map_err(|_| InternalError::ReadyQueueChannelClosed) + .map_err(|_| InternalError::InboundQueueChannelClosed) } } -/// A cloneable ready-queue receiver that reads from all three lanes. +/// A cloneable inbound-queue receiver that reads from all three lanes. /// /// Multiple consumers can clone this handle and concurrently receive from any lane; each entry is /// delivered to exactly one consumer. #[derive(Clone)] -pub struct ReadyQueueReceiverHandle { - task: Receiver>, - commit: Receiver>, - cleanup: Receiver>, +pub struct InboundQueueReceiverHandle { + task: Receiver>, + commit: Receiver>, + cleanup: Receiver>, } -impl ReadyQueueReceiverHandle { +impl InboundQueueReceiverHandle { /// Receives up to `max_items` regular task entries within a total time interval specified by /// `wait`. /// /// # Returns /// - /// The ready queue entries received from the ready queue, up to `max_items`. + /// The inbound queue entries received from the inbound queue, up to `max_items`. /// /// # Errors /// @@ -236,7 +236,7 @@ impl ReadyQueueReceiverHandle { &self, max_items: usize, wait: Duration, - ) -> Result>, InternalError> { + ) -> Result>, InternalError> { recv_batch(&self.task, max_items, wait).await } @@ -245,7 +245,7 @@ impl ReadyQueueReceiverHandle { /// /// # Returns /// - /// The ready queue entries received from the ready queue, up to `max_items`. + /// The inbound queue entries received from the inbound queue, up to `max_items`. /// /// # Errors /// @@ -256,7 +256,7 @@ impl ReadyQueueReceiverHandle { &self, max_items: usize, wait: Duration, - ) -> Result>, InternalError> { + ) -> Result>, InternalError> { recv_batch(&self.commit, max_items, wait).await } @@ -265,7 +265,7 @@ impl ReadyQueueReceiverHandle { /// /// # Returns /// - /// The ready queue entries received from the ready queue, up to `max_items`. + /// The inbound queue entries received from the inbound queue, up to `max_items`. /// /// # Errors /// @@ -276,42 +276,42 @@ impl ReadyQueueReceiverHandle { &self, max_items: usize, wait: Duration, - ) -> Result>, InternalError> { + ) -> Result>, InternalError> { recv_batch(&self.cleanup, max_items, wait).await } } /// Factory function. /// -/// Creates a ready queue and returns its paired sender and receiver handles. +/// Creates an inbound queue and returns its paired sender and receiver handles. /// /// # Returns /// /// A pair on success, containing: /// -/// * The sender handle of the ready queue. -/// * The receiver handle of the ready queue. +/// * The sender handle of the inbound queue. +/// * The receiver handle of the inbound queue. /// /// # Errors /// /// Returns an error if: /// -/// * Forwards [`ReadyQueueConfig::validate`]'s return values on failure. -pub fn create_ready_queue( - config: &ReadyQueueConfig, -) -> Result<(ReadyQueueSenderHandle, ReadyQueueReceiverHandle), InternalError> { +/// * Forwards [`InboundQueueConfig::validate`]'s return values on failure. +pub fn create_inbound_queue( + config: &InboundQueueConfig, +) -> Result<(InboundQueueSenderHandle, InboundQueueReceiverHandle), InternalError> { config.validate()?; let (task_tx, task_rx) = async_channel::bounded(config.task_capacity); let (commit_tx, commit_rx) = async_channel::bounded(config.commit_capacity); let (cleanup_tx, cleanup_rx) = async_channel::bounded(config.cleanup_capacity); - let sender = ReadyQueueSenderHandle { + let sender = InboundQueueSenderHandle { task: task_tx, commit: commit_tx, cleanup: cleanup_tx, }; - let receiver = ReadyQueueReceiverHandle { + let receiver = InboundQueueReceiverHandle { task: task_rx, commit: commit_rx, cleanup: cleanup_rx, @@ -340,19 +340,19 @@ const DEFAULT_CLEANUP_CAPACITY: usize = 1024; /// /// # Returns /// -/// The ready queue entries received from the lane, up to `max_items`. +/// The inbound queue entries received from the lane, up to `max_items`. /// /// # Errors /// /// Returns an error if: /// -/// * [`InternalError::ReadyQueueChannelClosed`] if the channel is closed. In a healthy storage +/// * [`InternalError::InboundQueueChannelClosed`] if the channel is closed. In a healthy storage /// service, the channel should only be closed when the service is shutting down. async fn recv_batch( - receiver: &Receiver>, + receiver: &Receiver>, max_items: usize, wait: Duration, -) -> Result>, InternalError> { +) -> Result>, InternalError> { if max_items == 0 { return Ok(Vec::new()); } @@ -364,7 +364,7 @@ async fn recv_batch( match result { Ok(entry) => entries.push(entry), Err(_) => { - return Err(InternalError::ReadyQueueChannelClosed); + return Err(InternalError::InboundQueueChannelClosed); } } } else { @@ -381,7 +381,7 @@ mod tests { #[tokio::test(flavor = "multi_thread")] async fn recv_returns_early_when_max_items_reached() -> anyhow::Result<()> { - let (sender, receiver) = create_ready_queue(&ReadyQueueConfig { + let (sender, receiver) = create_inbound_queue(&InboundQueueConfig { task_capacity: 1, commit_capacity: 1, cleanup_capacity: 1, @@ -427,7 +427,7 @@ mod tests { #[tokio::test(flavor = "multi_thread")] async fn recv_returns_empty_when_wait_elapses() -> anyhow::Result<()> { - let (_sender, receiver) = create_ready_queue(&ReadyQueueConfig { + let (_sender, receiver) = create_inbound_queue(&InboundQueueConfig { task_capacity: 1, commit_capacity: 1, cleanup_capacity: 1, diff --git a/components/spider-storage/src/lib.rs b/components/spider-storage/src/lib.rs index 6f68e5c8..4f9c0944 100644 --- a/components/spider-storage/src/lib.rs +++ b/components/spider-storage/src/lib.rs @@ -2,8 +2,8 @@ pub mod cache; mod config; pub mod db; pub mod grpc; +pub mod inbound_queue; pub mod job_submission; -pub mod ready_queue; pub mod state; pub mod task_instance_pool; diff --git a/components/spider-storage/src/state/job_cache.rs b/components/spider-storage/src/state/job_cache.rs index 298b641c..d2250f31 100644 --- a/components/spider-storage/src/state/job_cache.rs +++ b/components/spider-storage/src/state/job_cache.rs @@ -9,7 +9,7 @@ use crate::cache::error::CacheError; use crate::cache::error::InternalError; use crate::cache::job::SharedJobControlBlock; use crate::db::InternalJobOrchestration; -use crate::ready_queue::ReadyQueueSender; +use crate::inbound_queue::InboundQueueSender; use crate::state::StorageServerError; use crate::task_instance_pool::TaskInstancePoolConnector; @@ -20,23 +20,23 @@ use crate::task_instance_pool::TaskInstancePoolConnector; /// /// # Type Parameters /// -/// * `ReadyQueueSenderType` - The type of the ready queue sender. +/// * `InboundQueueSenderType` - The type of the inbound queue sender. /// * `DbConnectorType` - The type of the DB-layer connector. /// * `TaskInstancePoolConnectorType` - The type of the task instance pool connector. #[derive(Clone)] pub struct JobCache< - ReadyQueueSenderType: ReadyQueueSender, + InboundQueueSenderType: InboundQueueSender, DbConnectorType: InternalJobOrchestration, TaskInstancePoolConnectorType: TaskInstancePoolConnector, > { - jobs: SharedJobMap, + jobs: SharedJobMap, } impl< - ReadyQueueSenderType: ReadyQueueSender, + InboundQueueSenderType: InboundQueueSender, DbConnectorType: InternalJobOrchestration, TaskInstancePoolConnectorType: TaskInstancePoolConnector, -> JobCache +> JobCache { /// Creates a new empty job cache. #[must_use] @@ -57,7 +57,7 @@ impl< pub async fn insert( &self, jcb: SharedJobControlBlock< - ReadyQueueSenderType, + InboundQueueSenderType, DbConnectorType, TaskInstancePoolConnectorType, >, @@ -83,7 +83,11 @@ impl< &self, job_id: JobId, ) -> Option< - SharedJobControlBlock, + SharedJobControlBlock< + InboundQueueSenderType, + DbConnectorType, + TaskInstancePoolConnectorType, + >, > { self.jobs.read().await.get(&job_id).cloned() } @@ -97,7 +101,11 @@ impl< &self, job_id: JobId, ) -> Option< - SharedJobControlBlock, + SharedJobControlBlock< + InboundQueueSenderType, + DbConnectorType, + TaskInstancePoolConnectorType, + >, > { self.jobs.write().await.remove(&job_id) } @@ -115,7 +123,7 @@ impl< .count() } - /// Resends all ready tasks for every job in the cache to the ready queue. + /// Resends all ready tasks for every job in the cache to the inbound queue. /// /// # Errors /// @@ -131,23 +139,23 @@ impl< } impl< - ReadyQueueSenderType: ReadyQueueSender, + InboundQueueSenderType: InboundQueueSender, DbConnectorType: InternalJobOrchestration, TaskInstancePoolConnectorType: TaskInstancePoolConnector, -> Default for JobCache +> Default for JobCache { fn default() -> Self { Self::new() } } -type JobMap = HashMap< +type JobMap = HashMap< JobId, - SharedJobControlBlock, + SharedJobControlBlock, >; -type SharedJobMap = - Arc>>; +type SharedJobMap = + Arc>>; #[cfg(test)] mod tests { @@ -165,16 +173,19 @@ mod tests { use super::*; use crate::cache::error::InternalError; use crate::cache::job::SharedJobControlBlock; + use crate::inbound_queue::InboundQueueSender; use crate::job_submission::create_validated_submission; - use crate::ready_queue::ReadyQueueSender; use crate::state::test_utils::MockDbConnector; - use crate::state::test_utils::MockReadyQueueSender; + use crate::state::test_utils::MockInboundQueueSender; use crate::state::test_utils::MockTaskInstancePoolConnector; - async fn create_test_jcb( - job_id: JobId, - ) -> SharedJobControlBlock - { + type TestJcb = SharedJobControlBlock< + MockInboundQueueSender, + MockDbConnector, + MockTaskInstancePoolConnector, + >; + + async fn create_test_jcb(job_id: JobId) -> TestJcb { let bytes_type = DataTypeDescriptor::Value(ValueTypeDescriptor::bytes()); let mut submitted = SubmittedTaskGraph::new(None, None).expect("task graph creation should succeed"); @@ -197,7 +208,7 @@ mod tests { job_id, spider_core::types::id::ResourceGroupId::random(), job_submission, - MockReadyQueueSender, + MockInboundQueueSender, MockDbConnector::default(), MockTaskInstancePoolConnector, ) @@ -207,8 +218,11 @@ mod tests { #[tokio::test] async fn job_cache_insert_and_get() -> anyhow::Result<()> { - let cache: JobCache = - JobCache::new(); + let cache: JobCache< + MockInboundQueueSender, + MockDbConnector, + MockTaskInstancePoolConnector, + > = JobCache::new(); let job_id = JobId::random(); let jcb = create_test_jcb(job_id).await; @@ -221,8 +235,11 @@ mod tests { #[tokio::test] async fn job_cache_remove_returns_inserted_jcb() -> anyhow::Result<()> { - let cache: JobCache = - JobCache::new(); + let cache: JobCache< + MockInboundQueueSender, + MockDbConnector, + MockTaskInstancePoolConnector, + > = JobCache::new(); let job_id = JobId::random(); let jcb = create_test_jcb(job_id).await; @@ -238,8 +255,11 @@ mod tests { #[tokio::test] async fn job_cache_remove_batch_removes_existing_jobs_once() -> anyhow::Result<()> { - let cache: JobCache = - JobCache::new(); + let cache: JobCache< + MockInboundQueueSender, + MockDbConnector, + MockTaskInstancePoolConnector, + > = JobCache::new(); let first_job_id = JobId::random(); let second_job_id = JobId::random(); let missing_job_id = JobId::random(); @@ -268,8 +288,11 @@ mod tests { #[tokio::test] async fn job_cache_get_returns_none_for_nonexistent_job() -> anyhow::Result<()> { - let cache: JobCache = - JobCache::new(); + let cache: JobCache< + MockInboundQueueSender, + MockDbConnector, + MockTaskInstancePoolConnector, + > = JobCache::new(); let job_id = JobId::random(); let result = cache.get(job_id).await; @@ -282,8 +305,11 @@ mod tests { #[tokio::test] async fn job_cache_insert_duplicate_returns_error() -> anyhow::Result<()> { - let cache: JobCache = - JobCache::new(); + let cache: JobCache< + MockInboundQueueSender, + MockDbConnector, + MockTaskInstancePoolConnector, + > = JobCache::new(); let job_id = JobId::random(); let jcb1 = create_test_jcb(job_id).await; @@ -314,7 +340,7 @@ mod tests { use tokio_util::task::TaskTracker; let cache: Arc< - JobCache, + JobCache, > = Arc::new(JobCache::new()); let tracker = TaskTracker::new(); @@ -345,14 +371,14 @@ mod tests { tracker.wait().await; } - /// A tracking ready queue sender that records the number of calls. + /// A tracking inbound queue sender that records the number of calls. #[derive(Clone, Default)] - struct TrackingReadyQueueSender { + struct TrackingInboundQueueSender { call_count: Arc, } #[async_trait::async_trait] - impl ReadyQueueSender for TrackingReadyQueueSender { + impl InboundQueueSender for TrackingInboundQueueSender { async fn send_task_ready( &self, _rg_id: spider_core::types::id::ResourceGroupId, @@ -388,7 +414,7 @@ mod tests { #[tokio::test] async fn job_cache_resend_ready_tasks_sends_for_running_job() -> anyhow::Result<()> { let call_count: Arc = Arc::default(); - let sender = TrackingReadyQueueSender { + let sender = TrackingInboundQueueSender { call_count: Arc::clone(&call_count), }; @@ -425,7 +451,7 @@ mod tests { call_count.store(0, std::sync::atomic::Ordering::Relaxed); let cache: JobCache< - TrackingReadyQueueSender, + TrackingInboundQueueSender, MockDbConnector, MockTaskInstancePoolConnector, > = JobCache::new(); diff --git a/components/spider-storage/src/state/job_cache_gc.rs b/components/spider-storage/src/state/job_cache_gc.rs index c75ef8c2..0f5fb84d 100644 --- a/components/spider-storage/src/state/job_cache_gc.rs +++ b/components/spider-storage/src/state/job_cache_gc.rs @@ -14,7 +14,7 @@ use tokio_util::sync::CancellationToken; use crate::cache::error::InternalError; use crate::db::InternalJobOrchestration; -use crate::ready_queue::ReadyQueueSender; +use crate::inbound_queue::InboundQueueSender; use crate::state::JobCache; use crate::task_instance_pool::TaskInstancePoolConnector; @@ -88,7 +88,7 @@ impl JobCacheGcHandle { /// /// # Type Parameters /// -/// * `ReadyQueueSenderType` - The type of the ready queue sender required by the job cache. +/// * `InboundQueueSenderType` - The type of the inbound queue sender required by the job cache. /// * `DbConnectorType` - The type of the DB-layer connector required by the job cache. /// * `TaskInstancePoolConnectorType` - The type of the task instance pool connector required by the /// job cache. @@ -106,11 +106,11 @@ impl JobCacheGcHandle { /// /// * Forwards [`JobCacheGcConfig::validate`]'s return values on failure. pub fn create_job_cache_gc< - ReadyQueueSenderType: ReadyQueueSender + 'static, + InboundQueueSenderType: InboundQueueSender + 'static, DbConnectorType: InternalJobOrchestration + 'static, TaskInstancePoolConnectorType: TaskInstancePoolConnector + 'static, >( - job_cache: JobCache, + job_cache: JobCache, cancellation_token: CancellationToken, config: &JobCacheGcConfig, ) -> Result<(JobCacheGcHandle, JoinHandle>), InternalError> { @@ -137,27 +137,27 @@ struct TerminatedJob { } struct JobCacheGc< - ReadyQueueSenderType: ReadyQueueSender, + InboundQueueSenderType: InboundQueueSender, DbConnectorType: InternalJobOrchestration, TaskInstancePoolConnectorType: TaskInstancePoolConnector, > { - job_cache: JobCache, + job_cache: JobCache, terminated_jobs: VecDeque, terminated_job_retention: Duration, receiver: UnboundedReceiver, } impl< - ReadyQueueSenderType: ReadyQueueSender, + InboundQueueSenderType: InboundQueueSender, DbConnectorType: InternalJobOrchestration, TaskInstancePoolConnectorType: TaskInstancePoolConnector, -> JobCacheGc +> JobCacheGc { /// # Returns /// /// A new [`JobCacheGc`] actor over the given cache and message receiver. const fn new( - job_cache: JobCache, + job_cache: JobCache, terminated_job_retention: Duration, receiver: UnboundedReceiver, ) -> Self { @@ -255,11 +255,11 @@ mod tests { use super::JobCacheGcConfig; use crate::state::JobCache; use crate::state::test_utils::MockDbConnector; - use crate::state::test_utils::MockReadyQueueSender; + use crate::state::test_utils::MockInboundQueueSender; use crate::state::test_utils::MockTaskInstancePoolConnector; type TestJobCache = - JobCache; + JobCache; #[test] fn config_rejects_zero_values() { diff --git a/components/spider-storage/src/state/runtime.rs b/components/spider-storage/src/state/runtime.rs index 53ad4c41..902b38b2 100644 --- a/components/spider-storage/src/state/runtime.rs +++ b/components/spider-storage/src/state/runtime.rs @@ -11,10 +11,10 @@ use crate::config::DatabaseConfig; use crate::db::DbStorage; use crate::db::MariaDbStorageConnector; use crate::db::SessionManagement; -use crate::ready_queue::ReadyQueueConfig; -use crate::ready_queue::ReadyQueueSender; -use crate::ready_queue::ReadyQueueSenderHandle; -use crate::ready_queue::create_ready_queue; +use crate::inbound_queue::InboundQueueConfig; +use crate::inbound_queue::InboundQueueSender; +use crate::inbound_queue::InboundQueueSenderHandle; +use crate::inbound_queue::create_inbound_queue; use crate::state::JobCache; use crate::state::JobCacheGcConfig; use crate::state::ServiceState; @@ -31,7 +31,7 @@ use crate::task_instance_pool::create_task_instance_pool; pub struct RuntimeConfig { pub db: DatabaseConfig, #[serde(default)] - pub ready_queue: ReadyQueueConfig, + pub inbound_queue: InboundQueueConfig, #[serde(default)] pub task_instance_pool: TaskInstancePoolConfig, #[serde(default)] @@ -42,16 +42,16 @@ pub struct RuntimeConfig { /// /// # Type Parameters /// -/// * `ReadyQueueSenderType` - The ready queue sender type. +/// * `InboundQueueSenderType` - The inbound queue sender type. /// * `DbConnectorType` - The database connector type. /// * `TaskInstancePoolConnectorType` - The task instance pool connector type. pub struct Runtime< - ReadyQueueSenderType: ReadyQueueSender + 'static, + InboundQueueSenderType: InboundQueueSender + 'static, DbConnectorType: DbStorage + 'static, TaskInstancePoolConnectorType: TaskInstancePoolConnector + 'static, > { service_state: - ServiceState, + ServiceState, cancellation_token: CancellationToken, task_instance_pool_join_handle: JoinHandle>, job_cache_gc_join_handle: JoinHandle>, @@ -59,10 +59,10 @@ pub struct Runtime< } impl< - ReadyQueueSenderType: ReadyQueueSender + 'static, + InboundQueueSenderType: InboundQueueSender + 'static, DbConnectorType: DbStorage + 'static, TaskInstancePoolConnectorType: TaskInstancePoolConnector + 'static, -> Runtime +> Runtime { /// Stops the runtime. /// @@ -121,7 +121,7 @@ impl< #[must_use] pub fn get_service_state( &self, - ) -> ServiceState { + ) -> ServiceState { self.service_state.clone() } } @@ -141,13 +141,13 @@ impl< /// /// * Forwards [`MariaDbStorageConnector::connect`]'s return values on failure. /// * Forwards [`create_task_instance_pool`]'s return values on failure. -/// * Forwards [`create_ready_queue`]'s return values on failure. +/// * Forwards [`create_inbound_queue`]'s return values on failure. /// * Forwards [`create_job_cache_gc`]'s return values on failure. pub async fn create_runtime( config: &RuntimeConfig, ) -> Result< ( - Runtime, + Runtime, CancellationToken, ), StorageServerError, @@ -155,10 +155,10 @@ pub async fn create_runtime( let cancellation_token = CancellationToken::new(); let db = MariaDbStorageConnector::connect(&config.db).await?; let session_id = db.session_id(); - let (ready_queue_sender, ready_queue_receiver) = - create_ready_queue(&config.ready_queue).map_err(CacheError::from)?; + let (inbound_queue_sender, inbound_queue_receiver) = + create_inbound_queue(&config.inbound_queue).map_err(CacheError::from)?; let (task_instance_pool_connector, task_instance_pool_join_handle) = create_task_instance_pool( - ready_queue_sender.clone(), + inbound_queue_sender.clone(), db.clone(), cancellation_token.clone(), &config.task_instance_pool, @@ -167,7 +167,7 @@ pub async fn create_runtime( let job_cache = recover_job_cache( &db, - ready_queue_sender.clone(), + inbound_queue_sender.clone(), task_instance_pool_connector.clone(), ) .await?; @@ -181,8 +181,8 @@ pub async fn create_runtime( db, session_id, job_cache, - ready_queue_sender, - ready_queue_receiver, + inbound_queue_sender, + inbound_queue_receiver, task_instance_pool_connector, job_cache_gc_handle, cancellation_token: cancellation_token.clone(), @@ -216,15 +216,15 @@ const STOP_BACKGROUND_TASKS_TIMEOUT_SEC: u64 = 30; /// * Forwards [`SharedJobControlBlock::recover`]'s return values on failure. /// * Forwards [`JobCache::insert`]'s return values on failure. async fn recover_job_cache< - ReadyQueueSenderType: ReadyQueueSender, + InboundQueueSenderType: InboundQueueSender, DbConnectorType: DbStorage, TaskInstancePoolConnectorType: TaskInstancePoolConnector, >( db: &DbConnectorType, - ready_queue_sender: ReadyQueueSenderType, + inbound_queue_sender: InboundQueueSenderType, task_instance_pool_connector: TaskInstancePoolConnectorType, ) -> Result< - JobCache, + JobCache, StorageServerError, > { let job_cache = JobCache::new(); @@ -233,7 +233,7 @@ async fn recover_job_cache< let state = recoverable_job.state; let jcb = SharedJobControlBlock::recover( recoverable_job, - ready_queue_sender.clone(), + inbound_queue_sender.clone(), db.clone(), task_instance_pool_connector.clone(), ) @@ -258,9 +258,9 @@ mod tests { use super::*; use crate::cache::error::InternalError; use crate::db::SessionManagement; - use crate::ready_queue::ReadyQueueConfig; - use crate::ready_queue::ReadyQueueSenderHandle; - use crate::ready_queue::create_ready_queue; + use crate::inbound_queue::InboundQueueConfig; + use crate::inbound_queue::InboundQueueSenderHandle; + use crate::inbound_queue::create_inbound_queue; use crate::state::JobCache; use crate::state::ServiceState; use crate::state::ServiceStateParams; @@ -269,7 +269,7 @@ mod tests { use crate::state::test_utils::MockTaskInstancePoolConnector; type TestServerRuntime = - Runtime; + Runtime; fn create_test_runtime( cancellation_token: CancellationToken, @@ -279,7 +279,7 @@ mod tests { let db = MockDbConnector::default(); let session_id = db.session_id(); let (sender, receiver) = - create_ready_queue(&ReadyQueueConfig::default()).expect("ready queue creation"); + create_inbound_queue(&InboundQueueConfig::default()).expect("inbound queue creation"); let job_cache = JobCache::new(); let (job_cache_gc_handle, job_cache_gc_join_handle) = create_job_cache_gc( job_cache.clone(), @@ -291,8 +291,8 @@ mod tests { db, session_id, job_cache, - ready_queue_sender: sender, - ready_queue_receiver: receiver, + inbound_queue_sender: sender, + inbound_queue_receiver: receiver, task_instance_pool_connector: MockTaskInstancePoolConnector, job_cache_gc_handle, cancellation_token: cancellation_token.clone(), diff --git a/components/spider-storage/src/state/service.rs b/components/spider-storage/src/state/service.rs index 25228398..eb1c4275 100644 --- a/components/spider-storage/src/state/service.rs +++ b/components/spider-storage/src/state/service.rs @@ -23,12 +23,12 @@ use crate::cache::error::CacheError; use crate::cache::error::InternalError; use crate::cache::job::SharedJobControlBlock; use crate::db::DbStorage; +use crate::inbound_queue::CleanupTaskMarker; +use crate::inbound_queue::CommitTaskMarker; +use crate::inbound_queue::InboundQueueEntry; +use crate::inbound_queue::InboundQueueReceiverHandle; +use crate::inbound_queue::InboundQueueSender; use crate::job_submission::ValidatedJobSubmission; -use crate::ready_queue::CleanupTaskMarker; -use crate::ready_queue::CommitTaskMarker; -use crate::ready_queue::ReadyQueueEntry; -use crate::ready_queue::ReadyQueueReceiverHandle; -use crate::ready_queue::ReadyQueueSender; use crate::state::JobCache; use crate::state::JobCacheGcHandle; use crate::state::StorageServerError; @@ -40,19 +40,19 @@ use crate::task_instance_pool::TaskInstancePoolConnector; /// /// # Type Parameters /// -/// * `ReadyQueueSenderType` - The type of the ready queue sender. +/// * `InboundQueueSenderType` - The type of the inbound queue sender. /// * `DbConnectorType` - The type of the DB-layer connector. /// * `TaskInstancePoolConnectorType` - The type of the task instance pool connector. pub struct ServiceStateParams< - ReadyQueueSenderType: ReadyQueueSender + 'static, + InboundQueueSenderType: InboundQueueSender + 'static, DbConnectorType: DbStorage + 'static, TaskInstancePoolConnectorType: TaskInstancePoolConnector + 'static, > { pub db: DbConnectorType, pub session_id: SessionId, - pub job_cache: JobCache, - pub ready_queue_sender: ReadyQueueSenderType, - pub ready_queue_receiver: ReadyQueueReceiverHandle, + pub job_cache: JobCache, + pub inbound_queue_sender: InboundQueueSenderType, + pub inbound_queue_receiver: InboundQueueReceiverHandle, pub task_instance_pool_connector: TaskInstancePoolConnectorType, pub job_cache_gc_handle: JobCacheGcHandle, pub cancellation_token: CancellationToken, @@ -65,25 +65,25 @@ pub struct ServiceStateParams< /// /// # Type Parameters /// -/// * `ReadyQueueSenderType` - The type of the ready queue sender. +/// * `InboundQueueSenderType` - The type of the inbound queue sender. /// * `DbConnectorType` - The type of the DB-layer connector. /// * `TaskInstancePoolConnectorType` - The type of the task instance pool connector. #[derive(Clone)] pub struct ServiceState< - ReadyQueueSenderType: ReadyQueueSender + 'static, + InboundQueueSenderType: InboundQueueSender + 'static, DbConnectorType: DbStorage + 'static, TaskInstancePoolConnectorType: TaskInstancePoolConnector + 'static, > { inner: Arc< - ServiceStateInner, + ServiceStateInner, >, } impl< - ReadyQueueSenderType: ReadyQueueSender + 'static, + InboundQueueSenderType: InboundQueueSender + 'static, DbConnectorType: DbStorage + 'static, TaskInstancePoolConnectorType: TaskInstancePoolConnector + 'static, -> ServiceState +> ServiceState { /// Factory function. /// @@ -93,7 +93,7 @@ impl< #[must_use] pub fn new( params: ServiceStateParams< - ReadyQueueSenderType, + InboundQueueSenderType, DbConnectorType, TaskInstancePoolConnectorType, >, @@ -102,8 +102,8 @@ impl< db, session_id, job_cache, - ready_queue_sender, - ready_queue_receiver, + inbound_queue_sender, + inbound_queue_receiver, task_instance_pool_connector, job_cache_gc_handle, cancellation_token, @@ -113,8 +113,8 @@ impl< db, session_id, job_cache, - ready_queue_sender, - ready_queue_receiver, + inbound_queue_sender, + inbound_queue_receiver, task_instance_pool_connector, job_cache_gc_handle, has_previous_scheduler_connection: tokio::sync::Mutex::new(false), @@ -173,7 +173,7 @@ impl< job_id, resource_group_id, job_submission, - self.inner.ready_queue_sender.clone(), + self.inner.inbound_queue_sender.clone(), self.inner.db.clone(), self.inner.task_instance_pool_connector.clone(), ) @@ -307,7 +307,7 @@ impl< Ok(self.inner.db.get_error(job_id).await?) } - /// Resends ready tasks for all jobs in the cache to the ready queue. + /// Resends ready tasks for all jobs in the cache to the inbound queue. /// /// # Errors /// @@ -567,71 +567,71 @@ impl< .map_err(StorageServerError::from) } - /// Polls the ready queue for task entries. + /// Polls the inbound queue for task entries. /// /// # Returns /// - /// Up to `max_tasks` ready queue entries received within the `wait` duration on success. + /// Up to `max_tasks` inbound queue entries received within the `wait` duration on success. /// /// # Errors /// /// Returns an error if: /// - /// * Forwards [`ReadyQueueReceiverHandle::recv_tasks`]'s return values on failure. + /// * Forwards [`InboundQueueReceiverHandle::recv_tasks`]'s return values on failure. pub async fn poll_ready_tasks( &self, max_tasks: usize, wait: Duration, - ) -> Result>, StorageServerError> { + ) -> Result>, StorageServerError> { self.inner - .ready_queue_receiver + .inbound_queue_receiver .recv_tasks(max_tasks, wait) .await .map_err(|e| CacheError::Internal(e).into()) } - /// Polls the ready queue for commit-ready task entries. + /// Polls the inbound queue for commit-ready task entries. /// /// # Returns /// - /// Up to `max_tasks` commit-ready queue entries received within the `wait` duration on success. + /// Up to `max_tasks` commit-ready task entries received within the `wait` duration on success. /// /// # Errors /// /// Returns an error if: /// - /// * Forwards [`ReadyQueueReceiverHandle::recv_commits`]'s return values on failure. + /// * Forwards [`InboundQueueReceiverHandle::recv_commits`]'s return values on failure. pub async fn poll_commit_ready_tasks( &self, max_tasks: usize, wait: Duration, - ) -> Result>, StorageServerError> { + ) -> Result>, StorageServerError> { self.inner - .ready_queue_receiver + .inbound_queue_receiver .recv_commits(max_tasks, wait) .await .map_err(|e| CacheError::Internal(e).into()) } - /// Polls the ready queue for cleanup-ready task entries. + /// Polls the inbound queue for cleanup-ready task entries. /// /// # Returns /// - /// Up to `max_tasks` cleanup-ready queue entries received within the `wait` duration on + /// Up to `max_tasks` cleanup-ready task entries received within the `wait` duration on /// success. /// /// # Errors /// /// Returns an error if: /// - /// * Forwards [`ReadyQueueReceiverHandle::recv_cleanups`]'s return values on failure. + /// * Forwards [`InboundQueueReceiverHandle::recv_cleanups`]'s return values on failure. pub async fn poll_cleanup_ready_tasks( &self, max_tasks: usize, wait: Duration, - ) -> Result>, StorageServerError> { + ) -> Result>, StorageServerError> { self.inner - .ready_queue_receiver + .inbound_queue_receiver .recv_cleanups(max_tasks, wait) .await .map_err(|e| CacheError::Internal(e).into()) @@ -790,19 +790,19 @@ impl< /// /// # Type Parameters /// -/// * `ReadyQueueSenderType` - The type of the ready queue sender. +/// * `InboundQueueSenderType` - The type of the inbound queue sender. /// * `DbConnectorType` - The type of the DB-layer connector. /// * `TaskInstancePoolConnectorType` - The type of the task instance pool connector. struct ServiceStateInner< - ReadyQueueSenderType: ReadyQueueSender + 'static, + InboundQueueSenderType: InboundQueueSender + 'static, DbConnectorType: DbStorage + 'static, TaskInstancePoolConnectorType: TaskInstancePoolConnector + 'static, > { db: DbConnectorType, session_id: SessionId, - job_cache: JobCache, - ready_queue_sender: ReadyQueueSenderType, - ready_queue_receiver: ReadyQueueReceiverHandle, + job_cache: JobCache, + inbound_queue_sender: InboundQueueSenderType, + inbound_queue_receiver: InboundQueueReceiverHandle, task_instance_pool_connector: TaskInstancePoolConnectorType, job_cache_gc_handle: JobCacheGcHandle, has_previous_scheduler_connection: tokio::sync::Mutex, @@ -828,21 +828,21 @@ mod tests { use super::*; use crate::cache::job::SharedJobControlBlock; use crate::db::DbError; + use crate::inbound_queue::InboundQueueSenderHandle; use crate::job_submission::compress_job_inputs; use crate::job_submission::compress_task_graph; use crate::job_submission::create_validated_submission; - use crate::ready_queue::ReadyQueueSenderHandle; use crate::state::JobCacheGcHandle; use crate::state::StorageServerError; use crate::state::test_utils::MockDbConnector; - use crate::state::test_utils::MockReadyQueueSender; + use crate::state::test_utils::MockInboundQueueSender; use crate::state::test_utils::MockTaskInstancePoolConnector; type TestServiceState = - ServiceState; + ServiceState; - type TestServiceStateWithReadyQueue = - ServiceState; + type TestServiceStateWithInboundQueue = + ServiceState; const TEST_SESSION_ID: SessionId = 0; @@ -862,40 +862,40 @@ mod tests { db, session_id, job_cache: JobCache::new(), - ready_queue_sender: MockReadyQueueSender, - ready_queue_receiver: create_ready_queue_receiver(), + inbound_queue_sender: MockInboundQueueSender, + inbound_queue_receiver: create_inbound_queue_receiver(), task_instance_pool_connector: MockTaskInstancePoolConnector, job_cache_gc_handle: JobCacheGcHandle::new(tokio::sync::mpsc::unbounded_channel().0), cancellation_token: CancellationToken::new(), }) } - fn create_ready_queue_receiver() -> ReadyQueueReceiverHandle { - use crate::ready_queue::ReadyQueueConfig; - use crate::ready_queue::create_ready_queue; + fn create_inbound_queue_receiver() -> InboundQueueReceiverHandle { + use crate::inbound_queue::InboundQueueConfig; + use crate::inbound_queue::create_inbound_queue; let (_sender, receiver) = - create_ready_queue(&ReadyQueueConfig::default()).expect("ready queue creation"); + create_inbound_queue(&InboundQueueConfig::default()).expect("inbound queue creation"); receiver } - /// Creates a [`ServiceState`] backed by [`ReadyQueueReceiverHandle`]. + /// Creates a [`ServiceState`] backed by [`InboundQueueReceiverHandle`]. /// /// # Returns /// - /// A tuple of the service state and the ready queue sender handle on success. - fn create_test_service_with_ready_queue( + /// A tuple of the service state and the inbound queue sender handle on success. + fn create_test_service_with_inbound_queue( db: MockDbConnector, - ) -> (TestServiceStateWithReadyQueue, ReadyQueueSenderHandle) { - use crate::ready_queue::ReadyQueueConfig; - use crate::ready_queue::create_ready_queue; + ) -> (TestServiceStateWithInboundQueue, InboundQueueSenderHandle) { + use crate::inbound_queue::InboundQueueConfig; + use crate::inbound_queue::create_inbound_queue; let (sender, receiver) = - create_ready_queue(&ReadyQueueConfig::default()).expect("ready queue creation"); - let service = TestServiceStateWithReadyQueue::new(ServiceStateParams { + create_inbound_queue(&InboundQueueConfig::default()).expect("inbound queue creation"); + let service = TestServiceStateWithInboundQueue::new(ServiceStateParams { db, session_id: 0, job_cache: JobCache::new(), - ready_queue_sender: sender.clone(), - ready_queue_receiver: receiver, + inbound_queue_sender: sender.clone(), + inbound_queue_receiver: receiver, task_instance_pool_connector: MockTaskInstancePoolConnector, job_cache_gc_handle: JobCacheGcHandle::new(tokio::sync::mpsc::unbounded_channel().0), cancellation_token: CancellationToken::new(), @@ -941,10 +941,13 @@ mod tests { compress_job_inputs(&[]) } - async fn create_test_jcb( - job_id: JobId, - ) -> SharedJobControlBlock - { + type TestJcb = SharedJobControlBlock< + MockInboundQueueSender, + MockDbConnector, + MockTaskInstancePoolConnector, + >; + + async fn create_test_jcb(job_id: JobId) -> TestJcb { let task_graph = create_test_task_graph(); let inputs = vec![TaskInput::ValuePayload(vec![0u8; 4])]; let job_submission = create_validated_submission(task_graph, inputs); @@ -953,7 +956,7 @@ mod tests { job_id, ResourceGroupId::random(), job_submission, - MockReadyQueueSender, + MockInboundQueueSender, MockDbConnector::default(), MockTaskInstancePoolConnector, ) @@ -1435,8 +1438,8 @@ mod tests { db: MockDbConnector::default(), session_id: TEST_SESSION_ID, job_cache: JobCache::new(), - ready_queue_sender: MockReadyQueueSender, - ready_queue_receiver: create_ready_queue_receiver(), + inbound_queue_sender: MockInboundQueueSender, + inbound_queue_receiver: create_inbound_queue_receiver(), task_instance_pool_connector: MockTaskInstancePoolConnector, job_cache_gc_handle: JobCacheGcHandle::new(sender), cancellation_token: CancellationToken::new(), @@ -1462,8 +1465,8 @@ mod tests { db: MockDbConnector::default(), session_id: TEST_SESSION_ID, job_cache: JobCache::new(), - ready_queue_sender: MockReadyQueueSender, - ready_queue_receiver: create_ready_queue_receiver(), + inbound_queue_sender: MockInboundQueueSender, + inbound_queue_receiver: create_inbound_queue_receiver(), task_instance_pool_connector: MockTaskInstancePoolConnector, job_cache_gc_handle: JobCacheGcHandle::new(sender), cancellation_token: CancellationToken::new(), @@ -1512,8 +1515,8 @@ mod tests { db: MockDbConnector::default(), session_id: TEST_SESSION_ID, job_cache: JobCache::new(), - ready_queue_sender: MockReadyQueueSender, - ready_queue_receiver: create_ready_queue_receiver(), + inbound_queue_sender: MockInboundQueueSender, + inbound_queue_receiver: create_inbound_queue_receiver(), task_instance_pool_connector: MockTaskInstancePoolConnector, job_cache_gc_handle: JobCacheGcHandle::new(sender), cancellation_token: CancellationToken::new(), @@ -1662,9 +1665,9 @@ mod tests { } #[tokio::test] - async fn poll_ready_tasks_returns_entries_from_ready_queue() -> anyhow::Result<()> { + async fn poll_ready_tasks_returns_entries_from_inbound_queue() -> anyhow::Result<()> { const TASK_INDEX: TaskIndex = 0; - let (service, sender) = create_test_service_with_ready_queue(MockDbConnector::default()); + let (service, sender) = create_test_service_with_inbound_queue(MockDbConnector::default()); let rg_id = ResourceGroupId::random(); let job_id = JobId::random(); sender @@ -1675,7 +1678,7 @@ mod tests { let entries = service .poll_ready_tasks(10, Duration::from_millis(100)) .await?; - assert_eq!(entries.len(), 1, "should receive one ready queue entry"); + assert_eq!(entries.len(), 1, "should receive one inbound queue entry"); assert_eq!(entries[0].job_id, job_id); assert_eq!(entries[0].task_kind, TASK_INDEX); assert_eq!(entries[0].resource_group_id, rg_id); @@ -1684,20 +1687,20 @@ mod tests { #[tokio::test] async fn poll_ready_tasks_returns_empty_when_no_tasks() -> anyhow::Result<()> { - let (service, _sender) = create_test_service_with_ready_queue(MockDbConnector::default()); + let (service, _sender) = create_test_service_with_inbound_queue(MockDbConnector::default()); let entries = service .poll_ready_tasks(10, Duration::from_millis(10)) .await?; assert!( entries.is_empty(), - "should receive no entries from empty ready queue" + "should receive no entries from empty inbound queue" ); Ok(()) } #[tokio::test] - async fn poll_commit_ready_tasks_returns_entries_from_ready_queue() -> anyhow::Result<()> { - let (service, sender) = create_test_service_with_ready_queue(MockDbConnector::default()); + async fn poll_commit_ready_tasks_returns_entries_from_inbound_queue() -> anyhow::Result<()> { + let (service, sender) = create_test_service_with_inbound_queue(MockDbConnector::default()); let rg_id = ResourceGroupId::random(); let job_id = JobId::random(); sender @@ -1716,8 +1719,8 @@ mod tests { } #[tokio::test] - async fn poll_cleanup_ready_tasks_returns_entries_from_ready_queue() -> anyhow::Result<()> { - let (service, sender) = create_test_service_with_ready_queue(MockDbConnector::default()); + async fn poll_cleanup_ready_tasks_returns_entries_from_inbound_queue() -> anyhow::Result<()> { + let (service, sender) = create_test_service_with_inbound_queue(MockDbConnector::default()); let rg_id = ResourceGroupId::random(); let job_id = JobId::random(); sender @@ -1750,7 +1753,7 @@ mod tests { #[tokio::test] async fn register_scheduler_resends_ready_tasks_only_when_replacing_previous_scheduler() -> anyhow::Result<()> { - let (service, _sender) = create_test_service_with_ready_queue(MockDbConnector::default()); + let (service, _sender) = create_test_service_with_inbound_queue(MockDbConnector::default()); let (task_graph, inputs) = create_test_job_submission(); let job_id = service diff --git a/components/spider-storage/src/state/test_utils.rs b/components/spider-storage/src/state/test_utils.rs index 89f9cdc3..8f5b47b9 100644 --- a/components/spider-storage/src/state/test_utils.rs +++ b/components/spider-storage/src/state/test_utils.rs @@ -26,17 +26,17 @@ use crate::db::RecoverableJobContext; use crate::db::ResourceGroupManagement; use crate::db::SchedulerRegistrationManagement; use crate::db::SessionManagement; +use crate::inbound_queue::InboundQueueSender; use crate::job_submission::ValidatedJobSubmission; -use crate::ready_queue::ReadyQueueSender; use crate::task_instance_pool::TaskInstanceMetadata; use crate::task_instance_pool::TaskInstancePoolConnector; -/// A mock ready queue sender for testing. +/// A mock inbound queue sender for testing. #[derive(Clone, Default)] -pub struct MockReadyQueueSender; +pub struct MockInboundQueueSender; #[async_trait::async_trait] -impl ReadyQueueSender for MockReadyQueueSender { +impl InboundQueueSender for MockInboundQueueSender { async fn send_task_ready( &self, _rg_id: ResourceGroupId, diff --git a/components/spider-storage/src/task_instance_pool.rs b/components/spider-storage/src/task_instance_pool.rs index c7518386..302fe0ed 100644 --- a/components/spider-storage/src/task_instance_pool.rs +++ b/components/spider-storage/src/task_instance_pool.rs @@ -35,7 +35,7 @@ use crate::cache::error::InternalError; use crate::cache::task::SharedTaskControlBlock; use crate::cache::task::SharedTerminationTaskControlBlock; use crate::db::ExecutionManagerLivenessManagement; -use crate::ready_queue::ReadyQueueSender; +use crate::inbound_queue::InboundQueueSender; /// Configuration for a task instance pool actor. /// @@ -268,13 +268,13 @@ enum PoolMessage { /// /// # Type Parameters /// -/// * `ReadyQueueSenderType` - The ready queue sender implementation for re-enqueue operations. +/// * `InboundQueueSenderType` - The inbound queue sender implementation for re-enqueue operations. /// * `LivenessStoreType` - The execution manager liveness store implementation. struct TaskInstancePool< - ReadyQueueSenderType: ReadyQueueSender, + InboundQueueSenderType: InboundQueueSender, LivenessStoreType: ExecutionManagerLivenessManagement, > { - ready_queue_sender: ReadyQueueSenderType, + inbound_queue_sender: InboundQueueSenderType, execution_manager_liveness_store: LivenessStoreType, execution_manager_pool: HashSet, execution_manager_stale_after_sec: u64, @@ -282,8 +282,10 @@ struct TaskInstancePool< receiver: mpsc::Receiver, } -impl - TaskInstancePool +impl< + InboundQueueSenderType: InboundQueueSender, + LivenessStoreType: ExecutionManagerLivenessManagement, +> TaskInstancePool { /// Runs the coroutine loop, processing messages and GC timer ticks. /// @@ -349,7 +351,7 @@ impl Result<(), InternalError> { match metadata.task_id { TaskId::Index(task_index) => { - self.ready_queue_sender + self.inbound_queue_sender .send_task_ready( metadata.resource_group_id, metadata.job_id, @@ -466,12 +468,12 @@ impl { - self.ready_queue_sender + self.inbound_queue_sender .send_commit_ready(metadata.resource_group_id, metadata.job_id) .await } TaskId::Cleanup => { - self.ready_queue_sender + self.inbound_queue_sender .send_cleanup_ready(metadata.resource_group_id, metadata.job_id) .await } @@ -483,7 +485,7 @@ impl( - ready_queue_sender: ReadyQueueSenderType, + inbound_queue_sender: InboundQueueSenderType, execution_manager_liveness_store: LivenessStoreType, cancellation_token: CancellationToken, config: &TaskInstancePoolConfig, @@ -519,7 +521,7 @@ pub fn create_task_instance_pool< let (sender, receiver) = mpsc::channel(config.message_channel_capacity); let pool = TaskInstancePool { - ready_queue_sender, + inbound_queue_sender, execution_manager_liveness_store, execution_manager_stale_after_sec: config.execution_manager_stale_cutoff_sec, instances: Vec::new(), @@ -608,19 +610,19 @@ mod tests { } #[derive(Clone, Debug, PartialEq, Eq)] - enum ReadyMessage { + enum InboundMessage { Task(JobId, usize), Commit(JobId), Cleanup(JobId), } #[derive(Clone, Default)] - struct MockReadyQueueSender { - sent_messages: Arc>>, + struct MockInboundQueueSender { + sent_messages: Arc>>, } #[async_trait] - impl ReadyQueueSender for MockReadyQueueSender { + impl InboundQueueSender for MockInboundQueueSender { async fn send_task_ready( &self, _rg_id: ResourceGroupId, @@ -631,7 +633,7 @@ mod tests { self.sent_messages .lock() .await - .push(ReadyMessage::Task(job_id, task_index)); + .push(InboundMessage::Task(job_id, task_index)); } Ok(()) } @@ -644,7 +646,7 @@ mod tests { self.sent_messages .lock() .await - .push(ReadyMessage::Commit(job_id)); + .push(InboundMessage::Commit(job_id)); Ok(()) } @@ -656,7 +658,7 @@ mod tests { self.sent_messages .lock() .await - .push(ReadyMessage::Cleanup(job_id)); + .push(InboundMessage::Cleanup(job_id)); Ok(()) } } @@ -749,13 +751,13 @@ mod tests { /// The `mpsc::Receiver` field is required by the struct but unused by these tests; the matching /// sender is dropped immediately. fn build_test_pool( - ready_queue_sender: MockReadyQueueSender, + inbound_queue_sender: MockInboundQueueSender, liveness_store: MockExecutionManagerLivenessManagement, execution_manager_stale_cutoff: Duration, - ) -> TaskInstancePool { + ) -> TaskInstancePool { let (_sender, receiver) = mpsc::channel(1); TaskInstancePool { - ready_queue_sender, + inbound_queue_sender, execution_manager_liveness_store: liveness_store, execution_manager_pool: HashSet::new(), execution_manager_stale_after_sec: execution_manager_stale_cutoff.as_secs(), @@ -771,7 +773,7 @@ mod tests { /// /// The job ID assigned to the task, so callers can match it against re-enqueue messages. async fn register_task_in_pool( - pool: &mut TaskInstancePool, + pool: &mut TaskInstancePool, tcb: &SharedTaskControlBlock, task_id: TaskId, task_instance_id: TaskInstanceId, @@ -800,10 +802,10 @@ mod tests { #[tokio::test] async fn dead_execution_manager_registration_triggers_recovery() { - let ready_queue_sender = MockReadyQueueSender::default(); + let inbound_queue_sender = MockInboundQueueSender::default(); let cancellation_token = CancellationToken::new(); let (pool, pool_join_handle) = create_task_instance_pool( - ready_queue_sender.clone(), + inbound_queue_sender.clone(), RejectAllLivenessStore, cancellation_token.clone(), &TaskInstancePoolConfig { @@ -834,9 +836,9 @@ mod tests { // Give the pool coroutine time to process the message. tokio::time::sleep(Duration::from_millis(100)).await; - let messages = ready_queue_sender.sent_messages.lock().await.clone(); + let messages = inbound_queue_sender.sent_messages.lock().await.clone(); assert!( - messages.contains(&ReadyMessage::Task(job_id, 0)), + messages.contains(&InboundMessage::Task(job_id, 0)), "task should be re-enqueued for dead EM, got: {messages:?}" ); cancellation_token.cancel(); @@ -849,11 +851,11 @@ mod tests { #[tokio::test] async fn valid_em_is_cached_and_subsequent_registrations_skip_verify() { - let ready_queue_sender = MockReadyQueueSender::default(); + let inbound_queue_sender = MockInboundQueueSender::default(); let liveness_store = MockExecutionManagerLivenessManagement::default(); let cancellation_token = CancellationToken::new(); let (pool, pool_join_handle) = create_task_instance_pool( - ready_queue_sender, + inbound_queue_sender, liveness_store.clone(), cancellation_token.clone(), &TaskInstancePoolConfig { @@ -909,7 +911,7 @@ mod tests { async fn spawned_pool_exits_when_cancelled() { let cancellation_token = CancellationToken::new(); let (_pool, pool_join_handle) = create_task_instance_pool( - MockReadyQueueSender::default(), + MockInboundQueueSender::default(), MockExecutionManagerLivenessManagement::default(), cancellation_token.clone(), &TaskInstancePoolConfig { @@ -931,10 +933,10 @@ mod tests { #[tokio::test] async fn spawned_pool_processes_registration_before_shutdown() { - let ready_queue_sender = MockReadyQueueSender::default(); + let inbound_queue_sender = MockInboundQueueSender::default(); let cancellation_token = CancellationToken::new(); let (pool, pool_join_handle) = create_task_instance_pool( - ready_queue_sender.clone(), + inbound_queue_sender.clone(), RejectAllLivenessStore, cancellation_token.clone(), &TaskInstancePoolConfig { @@ -969,9 +971,9 @@ mod tests { .expect("pool task should join successfully") .expect("pool task should return success"); - let messages = ready_queue_sender.sent_messages.lock().await.clone(); + let messages = inbound_queue_sender.sent_messages.lock().await.clone(); assert!( - messages.contains(&ReadyMessage::Task(job_id, 0)), + messages.contains(&InboundMessage::Task(job_id, 0)), "registration should be processed before shutdown, got: {messages:?}" ); } @@ -980,10 +982,10 @@ mod tests { async fn gc_removes_all_terminated_tasks() { const NUM_TASKS: usize = 10; - let ready_queue_sender = MockReadyQueueSender::default(); + let inbound_queue_sender = MockInboundQueueSender::default(); let liveness_store = MockExecutionManagerLivenessManagement::default(); let mut pool = build_test_pool( - ready_queue_sender.clone(), + inbound_queue_sender.clone(), liveness_store, Duration::from_mins(1), ); @@ -1013,7 +1015,7 @@ mod tests { "all terminated entries should be removed, remaining: {}", pool.instances.len() ); - let messages = ready_queue_sender.sent_messages.lock().await.clone(); + let messages = inbound_queue_sender.sent_messages.lock().await.clone(); assert!( messages.is_empty(), "terminated tasks should not be re-enqueued, got: {messages:?}" @@ -1024,10 +1026,10 @@ mod tests { async fn gc_re_enqueues_all_soft_timed_out_tasks_and_keeps_them() { const NUM_TASKS: usize = 10; - let ready_queue_sender = MockReadyQueueSender::default(); + let inbound_queue_sender = MockInboundQueueSender::default(); let liveness_store = MockExecutionManagerLivenessManagement::default(); let mut pool = build_test_pool( - ready_queue_sender.clone(), + inbound_queue_sender.clone(), liveness_store, Duration::from_mins(1), ); @@ -1037,7 +1039,7 @@ mod tests { // deadline = now - 900ms let registered_at = gc_starting_time - Duration::from_secs(1); - let mut expected_messages: Vec = Vec::new(); + let mut expected_messages: Vec = Vec::new(); for i in 0..NUM_TASKS { let tcb = build_single_task_tcb().await; let job_id = register_task_in_pool( @@ -1049,7 +1051,7 @@ mod tests { registered_at, ) .await; - expected_messages.push(ReadyMessage::Task(job_id, i)); + expected_messages.push(InboundMessage::Task(job_id, i)); } pool.run_gc_cycle_at(gc_starting_time) @@ -1068,7 +1070,7 @@ mod tests { entry.metadata.task_instance_id ); } - let messages = ready_queue_sender.sent_messages.lock().await.clone(); + let messages = inbound_queue_sender.sent_messages.lock().await.clone(); assert_eq!(messages.len(), expected_messages.len(), "got: {messages:?}"); for expected in &expected_messages { assert!( @@ -1082,17 +1084,17 @@ mod tests { async fn gc_re_enqueues_and_removes_all_tasks_for_dead_em() { const NUM_TASKS: usize = 10; - let ready_queue_sender = MockReadyQueueSender::default(); + let inbound_queue_sender = MockInboundQueueSender::default(); let liveness_store = MockExecutionManagerLivenessManagement::default(); let mut pool = build_test_pool( - ready_queue_sender.clone(), + inbound_queue_sender.clone(), liveness_store.clone(), Duration::from_mins(1), ); let em_id = ExecutionManagerId::random(); let now = SystemTime::now(); - let mut expected_messages: Vec = Vec::new(); + let mut expected_messages: Vec = Vec::new(); for i in 0..NUM_TASKS { let tcb = build_single_task_tcb().await; let job_id = register_task_in_pool( @@ -1104,7 +1106,7 @@ mod tests { now, ) .await; - expected_messages.push(ReadyMessage::Task(job_id, i)); + expected_messages.push(InboundMessage::Task(job_id, i)); } liveness_store @@ -1126,7 +1128,7 @@ mod tests { !pool.execution_manager_pool.contains(&em_id), "dead EM should be pruned from execution_manager_pool" ); - let messages = ready_queue_sender.sent_messages.lock().await.clone(); + let messages = inbound_queue_sender.sent_messages.lock().await.clone(); assert_eq!(messages.len(), expected_messages.len(), "got: {messages:?}"); for expected in &expected_messages { assert!( @@ -1140,10 +1142,10 @@ mod tests { async fn gc_removes_terminated_tasks_for_dead_em_without_re_enqueue() { const NUM_TASKS: usize = 10; - let ready_queue_sender = MockReadyQueueSender::default(); + let inbound_queue_sender = MockInboundQueueSender::default(); let liveness_store = MockExecutionManagerLivenessManagement::default(); let mut pool = build_test_pool( - ready_queue_sender.clone(), + inbound_queue_sender.clone(), liveness_store.clone(), Duration::from_mins(1), ); @@ -1183,7 +1185,7 @@ mod tests { !pool.execution_manager_pool.contains(&em_id), "dead EM should be pruned from execution_manager_pool" ); - let messages = ready_queue_sender.sent_messages.lock().await.clone(); + let messages = inbound_queue_sender.sent_messages.lock().await.clone(); assert!( messages.is_empty(), "terminated tasks should not be re-enqueued even with dead EM, got: {messages:?}" @@ -1198,10 +1200,10 @@ mod tests { // index 2: alive EM, healthy on-going -> kept, no re-enqueue // index 3: dead EM, terminated -> removed, no re-enqueue (terminal wins) // index 4: dead EM, on-going -> removed, re-enqueued - let ready_queue_sender = MockReadyQueueSender::default(); + let inbound_queue_sender = MockInboundQueueSender::default(); let liveness_store = MockExecutionManagerLivenessManagement::default(); let mut pool = build_test_pool( - ready_queue_sender.clone(), + inbound_queue_sender.clone(), liveness_store.clone(), Duration::from_mins(1), ); @@ -1286,10 +1288,10 @@ mod tests { "alive EM should remain in execution_manager_pool" ); - let messages = ready_queue_sender.sent_messages.lock().await.clone(); + let messages = inbound_queue_sender.sent_messages.lock().await.clone(); let expected = [ - ReadyMessage::Task(job_id_1, 1), - ReadyMessage::Task(job_id_4, 4), + InboundMessage::Task(job_id_1, 1), + InboundMessage::Task(job_id_4, 4), ]; assert_eq!(messages.len(), expected.len(), "got: {messages:?}"); for msg in &expected { diff --git a/components/spider-storage/tests/runtime_recovery_test.rs b/components/spider-storage/tests/runtime_recovery_test.rs index 870c53ee..e5410ad9 100644 --- a/components/spider-storage/tests/runtime_recovery_test.rs +++ b/components/spider-storage/tests/runtime_recovery_test.rs @@ -11,10 +11,10 @@ use spider_core::types::io::TaskOutputsSerializer; use spider_storage::cache::error::CacheError; use spider_storage::cache::error::StaleStateError; use spider_storage::db::ExternalJobOrchestration; -use spider_storage::ready_queue::CleanupTaskMarker; -use spider_storage::ready_queue::CommitTaskMarker; -use spider_storage::ready_queue::ReadyQueueConfig; -use spider_storage::ready_queue::ReadyQueueEntry; +use spider_storage::inbound_queue::CleanupTaskMarker; +use spider_storage::inbound_queue::CommitTaskMarker; +use spider_storage::inbound_queue::InboundQueueConfig; +use spider_storage::inbound_queue::InboundQueueEntry; use spider_storage::state::JobCacheGcConfig; use spider_storage::state::Runtime; use spider_storage::state::ServiceState; @@ -99,10 +99,10 @@ async fn restarted_storage_cache_recovers_commit_ready_job() -> anyhow::Result<( let recovered_service = recovered_runtime.get_service_state(); recovered_service.resend_ready_tasks().await?; - let ready_entries = recovered_service + let inbound_entries = recovered_service .poll_commit_ready_tasks(32, Duration::from_secs(1)) .await?; - let job_entries = find_entry_for_job(ready_entries, job_id); + let job_entries = find_entry_for_job(inbound_entries, job_id); assert_eq!(job_entries.len(), 1); assert_eq!(job_entries[0].task_kind, CommitTaskMarker); @@ -156,10 +156,10 @@ async fn restarted_storage_cache_recovers_cleanup_ready_job() -> anyhow::Result< let recovered_service = recovered_runtime.get_service_state(); recovered_service.resend_ready_tasks().await?; - let ready_entries = recovered_service + let inbound_entries = recovered_service .poll_cleanup_ready_tasks(32, Duration::from_secs(1)) .await?; - let job_entries = find_entry_for_job(ready_entries, job_id); + let job_entries = find_entry_for_job(inbound_entries, job_id); assert_eq!(job_entries.len(), 1); assert_eq!(job_entries[0].task_kind, CleanupTaskMarker); @@ -202,7 +202,7 @@ async fn restarted_storage_cache_recovers_cleanup_ready_job() -> anyhow::Result< fn create_runtime_config() -> RuntimeConfig { RuntimeConfig { db: create_mariadb_config(), - ready_queue: ReadyQueueConfig::default(), + inbound_queue: InboundQueueConfig::default(), task_instance_pool: TaskInstancePoolConfig::default(), job_cache_gc: JobCacheGcConfig::default(), } @@ -231,7 +231,7 @@ async fn restart_after_starting_job( ) -> anyhow::Result<( JobId, Runtime< - spider_storage::ready_queue::ReadyQueueSenderHandle, + spider_storage::inbound_queue::InboundQueueSenderHandle, spider_storage::db::MariaDbStorageConnector, spider_storage::task_instance_pool::TaskInstancePoolHandle, >, @@ -270,7 +270,7 @@ async fn restart_after_commit_ready( ) -> anyhow::Result<( JobId, Runtime< - spider_storage::ready_queue::ReadyQueueSenderHandle, + spider_storage::inbound_queue::InboundQueueSenderHandle, spider_storage::db::MariaDbStorageConnector, spider_storage::task_instance_pool::TaskInstancePoolHandle, >, @@ -279,19 +279,19 @@ async fn restart_after_commit_ready( let service = runtime.get_service_state(); service.resend_ready_tasks().await?; - let ready_entries = service.poll_ready_tasks(32, Duration::from_secs(1)).await?; - let job_entries = find_entry_for_job(ready_entries, job_id); + let inbound_entries = service.poll_ready_tasks(32, Duration::from_secs(1)).await?; + let job_entries = find_entry_for_job(inbound_entries, job_id); assert_eq!(job_entries.len(), 1); - let ready_entry = job_entries[0]; + let inbound_entry = job_entries[0]; let task_instance_id = - run_recovered_regular_task(&service, job_id, ready_entry.task_kind).await?; + run_recovered_regular_task(&service, job_id, inbound_entry.task_kind).await?; let state = service .succeed_task_instance( service.session_id(), job_id, task_instance_id, - ready_entry.task_kind, + inbound_entry.task_kind, serialized_single_output()?, ) .await?; @@ -325,7 +325,7 @@ async fn restart_after_cleanup_ready( ) -> anyhow::Result<( JobId, Runtime< - spider_storage::ready_queue::ReadyQueueSenderHandle, + spider_storage::inbound_queue::InboundQueueSenderHandle, spider_storage::db::MariaDbStorageConnector, spider_storage::task_instance_pool::TaskInstancePoolHandle, >, @@ -354,11 +354,11 @@ async fn restart_after_cleanup_ready( /// * Forwards [`register_job`]'s return values on failure. /// * Forwards [`ServiceState::start_job`]'s return values on failure. async fn register_and_start_job< - ReadyQueueSenderType: spider_storage::ready_queue::ReadyQueueSender, + InboundQueueSenderType: spider_storage::inbound_queue::InboundQueueSender, DbConnectorType: spider_storage::db::DbStorage, TaskInstancePoolConnectorType: spider_storage::task_instance_pool::TaskInstancePoolConnector, >( - service: &ServiceState, + service: &ServiceState, with_commit: bool, with_cleanup: bool, ) -> anyhow::Result { @@ -381,11 +381,11 @@ async fn register_and_start_job< /// * Forwards [`spider_core::task::TaskGraph::to_json`]'s return values on failure. /// * Forwards [`ServiceState::register_job`]'s return values on failure. async fn register_job< - ReadyQueueSenderType: spider_storage::ready_queue::ReadyQueueSender, + InboundQueueSenderType: spider_storage::inbound_queue::InboundQueueSender, DbConnectorType: spider_storage::db::DbStorage, TaskInstancePoolConnectorType: spider_storage::task_instance_pool::TaskInstancePoolConnector, >( - service: &ServiceState, + service: &ServiceState, with_commit: bool, with_cleanup: bool, ) -> anyhow::Result { @@ -416,11 +416,11 @@ async fn register_job< /// * Forwards [`ServiceState::register_execution_manager`]'s return values on failure. /// * Forwards [`ServiceState::create_task_instance`]'s return values on failure. async fn run_recovered_regular_task< - ReadyQueueSenderType: spider_storage::ready_queue::ReadyQueueSender, + InboundQueueSenderType: spider_storage::inbound_queue::InboundQueueSender, DbConnectorType: spider_storage::db::DbStorage, TaskInstancePoolConnectorType: spider_storage::task_instance_pool::TaskInstancePoolConnector, >( - service: &ServiceState, + service: &ServiceState, job_id: JobId, task_index: TaskIndex, ) -> anyhow::Result { @@ -454,15 +454,15 @@ async fn run_recovered_regular_task< /// * Forwards [`ServiceState::succeed_task_instance`]'s return values on failure. /// * Forwards [`TaskOutputsSerializer::deserialize`]'s return values on failure. async fn run_single_task_job_to_succeed< - ReadyQueueSenderType: spider_storage::ready_queue::ReadyQueueSender, + InboundQueueSenderType: spider_storage::inbound_queue::InboundQueueSender, DbConnectorType: spider_storage::db::DbStorage, TaskInstancePoolConnectorType: spider_storage::task_instance_pool::TaskInstancePoolConnector, >( - service: &ServiceState, + service: &ServiceState, job_id: JobId, ) -> anyhow::Result> { - let ready_entries = service.poll_ready_tasks(32, Duration::from_secs(1)).await?; - let job_entries = find_entry_for_job(ready_entries, job_id); + let inbound_entries = service.poll_ready_tasks(32, Duration::from_secs(1)).await?; + let job_entries = find_entry_for_job(inbound_entries, job_id); assert_eq!(job_entries.len(), 1); let task_index = job_entries[0].task_kind; @@ -491,11 +491,11 @@ async fn run_single_task_job_to_succeed< /// Panics if the registration succeeds or fails with an error other than /// [`StaleStateError::JobNoLongerRunning`]. async fn assert_regular_task_registration_rejected< - ReadyQueueSenderType: spider_storage::ready_queue::ReadyQueueSender, + InboundQueueSenderType: spider_storage::inbound_queue::InboundQueueSender, DbConnectorType: spider_storage::db::DbStorage, TaskInstancePoolConnectorType: spider_storage::task_instance_pool::TaskInstancePoolConnector, >( - service: &ServiceState, + service: &ServiceState, job_id: JobId, execution_manager_id: ExecutionManagerId, ) { @@ -528,11 +528,11 @@ async fn assert_regular_task_registration_rejected< /// * Forwards [`ServiceState::get_job_state`]'s return values. /// * Forwards [`ServiceState::get_job_outputs`]'s return values. async fn assert_job_outputs_on_success< - ReadyQueueSenderType: spider_storage::ready_queue::ReadyQueueSender, + InboundQueueSenderType: spider_storage::inbound_queue::InboundQueueSender, DbConnectorType: spider_storage::db::DbStorage, TaskInstancePoolConnectorType: spider_storage::task_instance_pool::TaskInstancePoolConnector, >( - service: &ServiceState, + service: &ServiceState, job_id: JobId, expected_outputs: &[TaskOutput], ) -> anyhow::Result<()> { @@ -543,15 +543,15 @@ async fn assert_job_outputs_on_success< Ok(()) } -/// Collects the ready-queue entries belonging to a job. +/// Collects the inbound-queue entries belonging to a job. /// /// # Returns /// -/// The matching ready-queue entries, preserving their original order. +/// The matching inbound-queue entries, preserving their original order. fn find_entry_for_job( - entries: Vec>, + entries: Vec>, job_id: JobId, -) -> Vec> { +) -> Vec> { entries .into_iter() .filter(|entry| entry.job_id == job_id) diff --git a/components/spider-storage/tests/scheduling_infra.rs b/components/spider-storage/tests/scheduling_infra.rs index eb912415..dc0adb6f 100644 --- a/components/spider-storage/tests/scheduling_infra.rs +++ b/components/spider-storage/tests/scheduling_infra.rs @@ -30,8 +30,8 @@ //! //! # Mock components //! -//! * [`MockReadyQueueSender`] -- backed by `async_channel` (true MPMC). Task-ready batches from the -//! JCB are flattened to one message per task index for maximum execution manager concurrency. +//! * [`MockInboundQueueSender`] -- backed by `async_channel` (true MPMC). Task-ready batches from +//! the JCB are flattened to one message per task index for maximum execution manager concurrency. //! * [`NoopDbConnector`] -- stateless stub returning appropriate state transitions based on //! commit/cleanup task presence. Generic over `DbConnectorType` so a real connector can be //! swapped in. @@ -41,7 +41,7 @@ //! //! Each execution manager owns a cloned [`EmContext`] (no shared `Arc` indirection -- all fields //! are cheaply cloneable via `Arc` or built-in `Clone`). Execution managers `tokio::select!` -//! between the ready-queue receiver and a done signal (`watch`). +//! between the inbound-queue receiver and a done signal (`watch`). //! //! **Failure injection**: 50% of first-seen tasks spawn 2 concurrent `tokio::spawn` coroutines //! that each independently register and fail a task instance. This exercises both the retry @@ -103,8 +103,8 @@ use spider_storage::db::ExternalJobOrchestration; use spider_storage::db::InternalJobOrchestration; use spider_storage::db::MariaDbStorageConnector; use spider_storage::db::RecoverableJobContext; +use spider_storage::inbound_queue::InboundQueueSender; use spider_storage::job_submission::ValidatedJobSubmission; -use spider_storage::ready_queue::ReadyQueueSender; use spider_storage::task_instance_pool::TaskInstanceMetadata; use spider_storage::task_instance_pool::TaskInstancePoolConnector; use tabled::Table; @@ -334,9 +334,9 @@ pub async fn run_workload( instrument_sender: Option, ) -> WorkloadResult { // Create mock components. - let (ready_sender, ready_receiver) = async_channel::unbounded::(); - let ready_queue_sender = MockReadyQueueSender { - sender: ready_sender, + let (inbound_sender, inbound_receiver) = async_channel::unbounded::(); + let inbound_queue_sender = MockInboundQueueSender { + sender: inbound_sender, }; let (db_connector, job_id, resource_group_id) = db_connector_factory(&job_submission).await; let task_instance_pool = MockTaskInstancePool::new(); @@ -346,7 +346,7 @@ pub async fn run_workload( job_id, resource_group_id, job_submission, - ready_queue_sender, + inbound_queue_sender, db_connector, task_instance_pool, ) @@ -368,7 +368,7 @@ pub async fn run_workload( let cleanup_count = Arc::new(AtomicUsize::new(0)); let ctx = EmContext { - receiver: ready_receiver, + receiver: inbound_receiver, jcb: jcb.clone(), execution_manager_id: ExecutionManagerId::random(), terminal_state_sender: terminal_state_sender.clone(), @@ -484,16 +484,16 @@ const INSTRUMENT_OUTPUT_DIR_ENV: &str = "SPIDER_TEST_INSTRUMENT_OUTPUT_DIR"; /// /// * `DbConnectorType` - The DB-layer connector implementation. type TestJcb = - SharedJobControlBlock; + SharedJobControlBlock; -/// A message sent through the mock ready queue. +/// A message sent through the mock inbound queue. /// /// Each message represents a single schedulable unit of work. Task-ready batches from the JCB are /// flattened into one message per task index so that execution managers receive tasks individually, /// enabling better concurrency across the execution manager pool. /// /// Since these tests run a single job at a time, messages do not carry a job ID. -enum ReadyMessage { +enum InboundMessage { /// A single task is ready to be scheduled. Task { task_index: TaskIndex }, @@ -504,18 +504,18 @@ enum ReadyMessage { Cleanup, } -/// A mock [`ReadyQueueSender`] backed by an [`async_channel::Sender`]. +/// A mock [`InboundQueueSender`] backed by an [`async_channel::Sender`]. /// /// Execution managers each hold a cloned [`async_channel::Receiver`] and can concurrently await /// messages without any mutex serialization. The `job_id` parameter in each trait method is /// discarded since only one job runs at a time. #[derive(Clone)] -struct MockReadyQueueSender { - sender: async_channel::Sender, +struct MockInboundQueueSender { + sender: async_channel::Sender, } #[async_trait] -impl ReadyQueueSender for MockReadyQueueSender { +impl InboundQueueSender for MockInboundQueueSender { async fn send_task_ready( &self, _resource_group_id: ResourceGroupId, @@ -524,9 +524,9 @@ impl ReadyQueueSender for MockReadyQueueSender { ) -> Result<(), InternalError> { for task_index in task_indices { self.sender - .send(ReadyMessage::Task { task_index }) + .send(InboundMessage::Task { task_index }) .await - .map_err(|_| InternalError::ReadyQueueChannelClosed)?; + .map_err(|_| InternalError::InboundQueueChannelClosed)?; } Ok(()) } @@ -537,9 +537,9 @@ impl ReadyQueueSender for MockReadyQueueSender { _job_id: JobId, ) -> Result<(), InternalError> { self.sender - .send(ReadyMessage::Commit) + .send(InboundMessage::Commit) .await - .map_err(|_| InternalError::ReadyQueueChannelClosed) + .map_err(|_| InternalError::InboundQueueChannelClosed) } async fn send_cleanup_ready( @@ -548,9 +548,9 @@ impl ReadyQueueSender for MockReadyQueueSender { _job_id: JobId, ) -> Result<(), InternalError> { self.sender - .send(ReadyMessage::Cleanup) + .send(InboundMessage::Cleanup) .await - .map_err(|_| InternalError::ReadyQueueChannelClosed) + .map_err(|_| InternalError::InboundQueueChannelClosed) } } @@ -742,9 +742,9 @@ impl InstrumentedJcb /// * `DbConnectorType` - The DB-layer connector implementation used by the JCB. #[derive(Clone)] struct EmContext { - /// The MPMC ready-queue receiver. Each clone can concurrently await messages without + /// The MPMC inbound-queue receiver. Each clone can concurrently await messages without /// serialization. - receiver: async_channel::Receiver, + receiver: async_channel::Receiver, /// The instrumented JCB under test. jcb: InstrumentedJcb, @@ -897,8 +897,8 @@ fn collect_instrument_table(receiver: mpsc::UnboundedReceiver) Table::new(rows).to_string() } -/// Runs a single execution manager that consumes [`ReadyMessage`]s from the shared queue and drives -/// task execution through the JCB. +/// Runs a single execution manager that consumes [`InboundMessage`]s from the shared queue and +/// drives task execution through the JCB. /// /// The execution manager loops until either the done signal fires or the receiver is closed /// (returns `None`). @@ -936,21 +936,21 @@ async fn run_execution_manager { + InboundMessage::Task { task_index } => { if let Err(e) = process_task(&ctx, &mut rng, task_index).await && !is_stale_state(&e) { bail!(e); } } - ReadyMessage::Commit => { + InboundMessage::Commit => { if let Err(e) = process_commit(&ctx).await && !is_stale_state(&e) { bail!(e); } } - ReadyMessage::Cleanup => { + InboundMessage::Cleanup => { if let Err(e) = process_cleanup(&ctx).await && !is_stale_state(&e) { diff --git a/tools/deployment/spider-helm/Chart.yaml b/tools/deployment/spider-helm/Chart.yaml index 46de908b..adee2226 100644 --- a/tools/deployment/spider-helm/Chart.yaml +++ b/tools/deployment/spider-helm/Chart.yaml @@ -2,7 +2,7 @@ apiVersion: "v2" name: "spider" description: "A Helm chart for the Spider Huntsman deployment" type: "application" -version: "0.1.11" +version: "0.1.12" appVersion: "0.1.0-dev" home: "https://github.com/y-scope/spider" sources: ["https://github.com/y-scope/spider"] diff --git a/tools/deployment/spider-helm/values.yaml b/tools/deployment/spider-helm/values.yaml index 41d52609..f839afee 100644 --- a/tools/deployment/spider-helm/values.yaml +++ b/tools/deployment/spider-helm/values.yaml @@ -75,13 +75,13 @@ spiderConfig: log_level: "INFO" port: 50051 runtime: - job_cache_gc: - gc_interval_sec: 30 - terminated_job_retention_sec: 300 - ready_queue: + inbound_queue: cleanup_capacity: 256 commit_capacity: 256 task_capacity: 1048576 + job_cache_gc: + gc_interval_sec: 30 + terminated_job_retention_sec: 300 task_instance_pool: execution_manager_stale_cutoff_sec: 60 gc_interval_sec: 30