wip

Author: artemist

subprojects/hydra-queue-runner/src/state/build.rs

@@ -25,7 +25,7 @@ pub struct Build {
     pub local_priority: i32,
     pub global_priority: AtomicI32,
 
-    toplevel: arc_swap::ArcSwapOption<Step>,
+    pub toplevel: arc_swap::ArcSwapOption<Step>,
     pub jobset: Arc<Jobset>,
 
     finished_in_db: AtomicBool,

subprojects/hydra-queue-runner/src/state/mod.rs

@@ -46,6 +46,8 @@ enum CreateStepResult {
 
 enum RealiseStepResult {
     None,
+    /// Created a new resolved BuildStep
+    Resolved,
     Valid(Arc<Machine>),
     MaybeCancelled,
     CachedFailure,
@@ -275,7 +277,7 @@ impl State {
         let drv = step_info.step.get_drv_path();
         let mut build_options = nix_utils::BuildOptions::new(None);
 
-        let build_id = {
+        let build = {
             let mut dependents = HashSet::new();
             let mut steps = HashSet::new();
             step_info.step.get_dependents(&mut dependents, &mut steps);
@@ -308,9 +310,11 @@ impl State {
             build_options
                 .set_max_silent_time(biggest_max_silent_time.unwrap_or(build.max_silent_time));
             build_options.set_build_timeout(biggest_build_timeout.unwrap_or(build.timeout));
-            build.id
+            build.clone()
         };
 
+        let build_id = build.id;
+
         let mut job = machine::Job::new(build_id, drv.to_owned());
         job.result.set_start_time_now();
         if self.check_cached_failure(step_info.step.clone()).await {
@@ -403,10 +407,11 @@ impl State {
             }
         };
 
-        let effective_drv = if let Some(resolved_path) = resolved {
+        if let Some(resolved_path) = resolved {
             tracing::info!("resolved CA derivation {drv} -> {resolved_path}");
 
             // Finish original step as "resolved".
+            step_info.step.set_finished(true);
             let mut resolved_result = RemoteBuild::new();
             resolved_result.step_status = BuildStatus::Resolved;
             resolved_result.set_start_time_now();
@@ -424,30 +429,68 @@ impl State {
             .await?;
 
             // Create DB step for the resolved drv under the same build.
-            let resolved_step_nr = {
-                let mut tx = db.begin_transaction().await?;
-                let nr = tx
-                    .create_build_step(
-                        Some(job.result.get_start_time_as_i32()?),
-                        build_id,
-                        &self.store.print_store_path(&resolved_path),
-                        step_info.step.get_system().as_deref(),
-                        machine.hostname.clone(),
-                        BuildStatus::Busy,
-                        None,
-                        None,
-                        vec![],
-                    )
-                    .await?;
+            let mut tx = db.begin_transaction().await?;
+            let nr = tx
+                .create_build_step(
+                    Some(job.result.get_start_time_as_i32()?),
+                    build_id,
+                    &self.store.print_store_path(&resolved_path),
+                    step_info.step.get_system().as_deref(),
+                    machine.hostname.clone(),
+                    BuildStatus::Busy,
+                    None,
+                    None,
+                    vec![],
+                )
+                .await?;
 
-                tx.set_resolved_to(build_id, step_nr, nr).await?;
-                tx.commit().await?;
-                nr
+            tx.set_resolved_to(build_id, step_nr, nr).await?;
+            tx.commit().await?;
+
+            // Actually schedule the step
+            let resolved_step = match self
+                .create_step(
+                    build.clone(),
+                    resolved_path,
+                    Some(build.clone()),
+                    None,
+                    Arc::new(parking_lot::RwLock::new(HashSet::new())),
+                    Arc::new(parking_lot::RwLock::new(HashSet::new())),
+                    Arc::new(parking_lot::RwLock::new(HashSet::new())),
+                )
+                .await
+            {
+                CreateStepResult::None => {
+                    return Err(anyhow::anyhow!("Could not create resolved build step"));
+                }
+                CreateStepResult::Valid(step) => step,
+                CreateStepResult::PreviousFailure(step) => step,
             };
-            job.step_nr = resolved_step_nr;
-            resolved_path
-        } else {
-            drv.to_owned()
+
+            resolved_step.make_runnable();
+
+            // Copy rdeps so dependents don't try to start too early
+            for rdep in step_info.step.clone_rdeps() {
+                if let Some(rdep) = rdep.upgrade() {
+                    resolved_step.make_rdep(&rdep);
+                }
+            }
+
+            // If we're the root of a build then we need to become the new root,
+            // lest the Step get garbage collected.
+            if *build
+                .toplevel
+                .compare_and_swap(&step_info.step, Some(resolved_step))
+                == Some(step_info.step.clone())
+            {
+                build.propagate_priorities();
+            }
+
+            // New steps runnable
+            self.trigger_dispatch();
+
+            // No more work to do, build will happen in another step
+            return Ok(RealiseStepResult::Resolved);
         };
 
         {
@@ -456,7 +499,7 @@ impl State {
             tx.commit().await?;
         }
         tracing::info!(
-            "Submitting build drv={effective_drv} on machine={} hostname={} build_id={build_id} step_nr={}",
+            "Submitting build drv={drv} on machine={} hostname={} build_id={build_id} step_nr={}",
             machine.id,
             machine.hostname,
             job.step_nr,
@@ -473,7 +516,7 @@ impl State {
         machine
             .build_drv(
                 job,
-                effective_drv,
+                drv.clone(),
                 &build_options,
                 // TODO: cleanup
                 if self.config.use_presigned_uploads() {

subprojects/hydra-queue-runner/src/state/queue.rs

@@ -504,6 +504,7 @@ impl Queues {
                         nr_waiting += 1;
                         nr_steps_waiting_all_queues += 1;
                     }
+                    Ok(crate::state::RealiseStepResult::Resolved) => {}
                     Ok(
                         crate::state::RealiseStepResult::MaybeCancelled
                         | crate::state::RealiseStepResult::CachedFailure,

subprojects/hydra-queue-runner/src/state/step.rs

@@ -351,6 +351,20 @@ impl Step {
             .store(state.deps.len() as u64, Ordering::Relaxed);
     }
 
+    pub fn make_rdep(self: &Arc<Self>, dep: &Arc<Self>) {
+        dep.add_dep(self.clone());
+        let mut state = self.state.write();
+        state.rdeps.push(Arc::downgrade(dep));
+        self.atomic_state
+            .rdeps_len
+            .store(state.rdeps.len() as u64, Ordering::Relaxed);
+    }
+
+    pub fn clone_rdeps(&self) -> Vec<Weak<Step>> {
+        let state = self.state.read();
+        state.rdeps.clone()
+    }
+
     pub fn add_referring_data(
         &self,
         referring_build: Option<&Arc<Build>>,