mingmwang commented on a change in pull request #1810:
URL: https://github.com/apache/arrow-datafusion/pull/1810#discussion_r804509614
##########
File path: ballista/rust/scheduler/src/lib.rs
##########
@@ -153,24 +154,18 @@ impl SchedulerServer {
.as_millis(),
policy,
scheduler_env,
- executors_client: Arc::new(RwLock::new(HashMap::new())),
+ executors_client,
}
}
+ pub async fn init(&self) -> Result<(), BallistaError> {
+ self.state.init().await?;
+
+ Ok(())
+ }
+
async fn schedule_job(&self, job_id: String) -> Result<(), BallistaError> {
- let alive_executors = self
- .state
- .get_alive_executors_metadata_within_one_minute()
- .await?;
- let alive_executors: HashMap<String, ExecutorMeta> = alive_executors
- .into_iter()
- .map(|e| (e.id.clone(), e))
- .collect();
- let available_executors =
self.state.get_available_executors_data().await?;
- let mut available_executors: Vec<ExecutorData> = available_executors
- .into_iter()
- .filter(|e| alive_executors.contains_key(&e.executor_id))
- .collect();
+ let mut available_executors =
self.state.get_available_executors_data();
// In case of there's no enough resources, reschedule the tasks of the
job
if available_executors.is_empty() {
Review comment:
Why spawn another future and just sleep ? I think you can just simply
sleep in the schedule_job() method if there is no enough resources.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscr...@arrow.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
