remove is_push_staged_scheduling method

systemxlabs · Jan 10, 2024 · 4ba8aa8 · 4ba8aa8
1 parent c10a0df
commit 4ba8aa8
Show file tree

Hide file tree

Showing 3 changed files with 12 additions and 29 deletions.
diff --git a/ballista/scheduler/src/config.rs b/ballista/scheduler/src/config.rs
@@ -89,11 +89,6 @@ impl SchedulerConfig {
         format!("{}:{}", self.external_host, self.bind_port)
     }
 
-    pub fn is_push_staged_scheduling(&self) -> bool {
-        // TODO lwz
-        true
-    }
-
     pub fn with_namespace(mut self, namespace: impl Into<String>) -> Self {
         self.namespace = namespace.into();
         self

diff --git a/ballista/scheduler/src/scheduler_server/mod.rs b/ballista/scheduler/src/scheduler_server/mod.rs
@@ -178,9 +178,7 @@ impl<T: 'static + AsLogicalPlan, U: 'static + AsExecutionPlan> SchedulerServer<T
         tasks_status: Vec<TaskStatus>,
     ) -> Result<()> {
         // We might receive buggy task updates from dead executors.
-        if self.state.config.is_push_staged_scheduling()
-            && self.state.executor_manager.is_dead_executor(executor_id)
-        {
+        if self.state.executor_manager.is_dead_executor(executor_id) {
             let error_msg = format!(
                 "Receive buggy tasks status from dead Executor {executor_id}, task status update ignored."
             );
@@ -306,11 +304,7 @@ impl<T: 'static + AsLogicalPlan, U: 'static + AsExecutionPlan> SchedulerServer<T
             .register_executor(metadata, executor_data)
             .await?;
 
-        // If we are using push-based scheduling then reserve this executors slots and send
-        // them for scheduling tasks.
-        if self.state.config.is_push_staged_scheduling() {
-            self.revive_offers().await?;
-        }
+        self.revive_offers().await?;
 
         Ok(())
     }

diff --git a/ballista/scheduler/src/scheduler_server/query_stage_scheduler.rs b/ballista/scheduler/src/scheduler_server/query_stage_scheduler.rs
@@ -141,11 +141,9 @@ impl<T: 'static + AsLogicalPlan, U: 'static + AsExecutionPlan>
 
                 info!("Job {} submitted", job_id);
 
-                if self.state.config.is_push_staged_scheduling() {
-                    event_sender
-                        .post_event(QueryStageSchedulerEvent::ReviveOffers)
-                        .await?;
-                }
+                event_sender
+                    .post_event(QueryStageSchedulerEvent::ReviveOffers)
+                    .await?;
             }
             QueryStageSchedulerEvent::JobPlanningFailed {
                 job_id,
@@ -257,23 +255,19 @@ impl<T: 'static + AsLogicalPlan, U: 'static + AsExecutionPlan>
                 );
 
                 let num_status = tasks_status.len();
-                if self.state.config.is_push_staged_scheduling() {
-                    self.state
-                        .executor_manager
-                        .unbind_tasks(vec![(executor_id.clone(), num_status as u32)])
-                        .await?;
-                }
+                self.state
+                    .executor_manager
+                    .unbind_tasks(vec![(executor_id.clone(), num_status as u32)])
+                    .await?;
                 match self
                     .state
                     .update_task_statuses(&executor_id, tasks_status)
                     .await
                 {
                     Ok(stage_events) => {
-                        if self.state.config.is_push_staged_scheduling() {
-                            event_sender
-                                .post_event(QueryStageSchedulerEvent::ReviveOffers)
-                                .await?;
-                        }
+                        event_sender
+                            .post_event(QueryStageSchedulerEvent::ReviveOffers)
+                            .await?;
 
                         for stage_event in stage_events {
                             event_sender.post_event(stage_event).await?;