Skip to content

chore: rework node scheduling #182

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
pchalamet merged 1 commit into from
Jun 14, 2025
Merged

chore: rework node scheduling #182

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
285 changes: 150 additions & 135 deletions src/Terrabuild/Core/Build.fs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -160,160 +160,174 @@ let run (options: ConfigOptions.Options) (cache: Cache.ICache) (api: Contracts.I
let allowRemoteCache = options.LocalOnly |> not
let homeDir = Cache.createHome()
let tmpDir = Cache.createTmp()
let force = options.Force
let retry = options.Retry

let nodeResults = Concurrent.ConcurrentDictionary<string, TaskRequest * TaskStatus>()
let restorables = Concurrent.ConcurrentDictionary<string, Restorable>()

let processNode (node: GraphDef.Node) =
let startedAt = DateTime.UtcNow
notification.NodeBuilding node
let processNode (maxCompletionChildren: DateTime) (node: GraphDef.Node) =
let cacheEntryId = GraphDef.buildCacheKey node

let projectDirectory =
match node.Project with
| FS.Directory projectDirectory -> projectDirectory
| FS.File projectFile -> projectFile |> FS.parentDirectory |> Option.get
| _ -> "."

// restore lazy dependencies
node.Dependencies |> Seq.iter (fun nodeId ->
match restorables.TryGetValue nodeId with
| true, restorable -> restorable.Restore()
| _ -> ())
let buildNode() =
let startedAt = DateTime.UtcNow

notification.NodeBuilding node

// restore lazy dependencies
node.Dependencies |> Seq.iter (fun nodeId ->
match restorables.TryGetValue nodeId with
| true, restorable -> restorable.Restore()
| _ -> ())

let beforeFiles =
if node.IsLeaf then IO.Snapshot.Empty
else IO.createSnapshot node.Outputs projectDirectory

let cacheEntry = cache.GetEntry true cacheEntryId
let lastStatusCode, stepLogs = execCommands node cacheEntry options projectDirectory homeDir tmpDir

// keep only new or modified files
let afterFiles = IO.createSnapshot node.Outputs projectDirectory
let newFiles = afterFiles - beforeFiles
let outputs = IO.copyFiles cacheEntry.Outputs projectDirectory newFiles

let successful = lastStatusCode = 0
let endedAt = DateTime.UtcNow
let summary =
{ Cache.TargetSummary.Project = node.Project
Cache.TargetSummary.Target = node.Target
Cache.TargetSummary.Operations = [ stepLogs |> List.ofSeq ]
Cache.TargetSummary.Outputs = outputs
Cache.TargetSummary.IsSuccessful = successful
Cache.TargetSummary.StartedAt = startedAt
Cache.TargetSummary.EndedAt = endedAt
Cache.TargetSummary.Duration = endedAt - startedAt
Cache.TargetSummary.Cache = node.Cache }

notification.NodeUploading node

// create an archive with new files
Log.Debug("{NodeId}: Building '{Project}/{Target}' with {Hash}", node.Id, node.Project, node.Target, node.TargetHash)
let files = cacheEntry.Complete summary
api |> Option.iter (fun api -> api.AddArtifact node.Project node.Target node.ProjectHash node.TargetHash files successful)

match lastStatusCode with
| 0 -> TaskStatus.Success endedAt
| _ -> TaskStatus.Failure (DateTime.UtcNow, $"{node.Id} failed with exit code {lastStatusCode}")

let restoreNode () =
notification.NodeScheduled node
let cacheEntryId = GraphDef.buildCacheKey node
match cache.TryGetSummaryOnly allowRemoteCache cacheEntryId with
| Some (_, summary) ->
let dependencies =
node.Dependencies |> Seq.choose (fun nodeId ->
match restorables.TryGetValue nodeId with
| true, restorable -> Some restorable
| _ -> None)
|> List.ofSeq

let beforeFiles =
if node.IsLeaf then IO.Snapshot.Empty
else IO.createSnapshot node.Outputs projectDirectory
let callback() =
if node.Managed then
notification.NodeDownloading node
match cache.TryGetSummary allowRemoteCache cacheEntryId with
| Some summary ->
Log.Debug("{NodeId} restoring '{Project}/{Target}' from cache from {Hash}", node.Id, node.Project, node.Target, node.TargetHash)
match summary.Outputs with
| Some outputs ->
let files = IO.enumerateFiles outputs
IO.copyFiles projectDirectory outputs files |> ignore
api |> Option.iter (fun api -> api.UseArtifact node.ProjectHash node.TargetHash)
| _ -> ()
notification.NodeCompleted node TaskRequest.Restore true
| _ ->
notification.NodeCompleted node TaskRequest.Restore false
raiseBugError $"Unable to download build output for {cacheEntryId} for node {node.Id}"

let cacheEntryId = GraphDef.buildCacheKey node
let cacheEntry = cache.GetEntry true cacheEntryId
let lastStatusCode, stepLogs = execCommands node cacheEntry options projectDirectory homeDir tmpDir

// keep only new or modified files
let afterFiles = IO.createSnapshot node.Outputs projectDirectory
let newFiles = afterFiles - beforeFiles
let outputs = IO.copyFiles cacheEntry.Outputs projectDirectory newFiles

let successful = lastStatusCode = 0
let endedAt = DateTime.UtcNow
let summary =
{ Cache.TargetSummary.Project = node.Project
Cache.TargetSummary.Target = node.Target
Cache.TargetSummary.Operations = [ stepLogs |> List.ofSeq ]
Cache.TargetSummary.Outputs = outputs
Cache.TargetSummary.IsSuccessful = successful
Cache.TargetSummary.StartedAt = startedAt
Cache.TargetSummary.EndedAt = endedAt
Cache.TargetSummary.Duration = endedAt - startedAt
Cache.TargetSummary.Cache = node.Cache }

notification.NodeUploading node

// create an archive with new files
Log.Debug("{NodeId}: Building '{Project}/{Target}' with {Hash}", node.Id, node.Project, node.Target, node.TargetHash)
let files = cacheEntry.Complete summary
api |> Option.iter (fun api -> api.AddArtifact node.Project node.Target node.ProjectHash node.TargetHash files successful)

match lastStatusCode with
| 0 -> TaskStatus.Success endedAt
| _ -> TaskStatus.Failure (DateTime.UtcNow, $"{node.Id} failed with exit code {lastStatusCode}")
let restorable = Restorable(callback, dependencies)
restorables.TryAdd(node.Id, restorable) |> ignore
if summary.IsSuccessful then TaskStatus.Success summary.EndedAt
else TaskStatus.Failure (summary.EndedAt, $"Restored node {node.Id} with a build in failure state")
| _ ->
TaskStatus.Failure (DateTime.UtcNow, $"Unable to download build output for {cacheEntryId} for node {node.Id}")

if node.Rebuild then
Log.Debug("{NodeId} must rebuild because force requested", node.Id)
TaskRequest.Build, buildNode()

let hub = Hub.Create(options.MaxConcurrency)
let rec schedule nodeId =
if nodeResults.TryAdd(nodeId, (TaskRequest.Build, TaskStatus.Pending)) then
let node = graph.Nodes[nodeId]
notification.NodeScheduled node
elif maxCompletionChildren = DateTime.MaxValue then
Log.Debug("{NodeId} must rebuild because child is rebuilding", node.Id)
TaskRequest.Build, buildNode()

let projectDirectory =
match node.Project with
| FS.Directory projectDirectory -> projectDirectory
| FS.File projectFile -> projectFile |> FS.parentDirectory |> Option.get
| _ -> "."
elif node.Cache <> Terrabuild.Extensibility.Cacheability.Never then
let cacheEntryId = GraphDef.buildCacheKey node
match cache.TryGetSummaryOnly allowRemoteCache cacheEntryId with
| Some (_, summary) ->
Log.Debug("{NodeId} has existing build summary", node.Id)

// task is failed and retry requested
if retry && not summary.IsSuccessful then
Log.Debug("{NodeId} must rebuild because node is failed and retry requested", node.Id)
TaskRequest.Build, buildNode()

let completionStatus =
if force || node.Rebuild then None
// task is cached
else
let cacheEntryId = GraphDef.buildCacheKey node
match cache.TryGetSummaryOnly allowRemoteCache cacheEntryId with
| Some (_, summary) ->
if retry && not summary.IsSuccessful then None
else
let dependencies =
node.Dependencies |> Seq.choose (fun nodeId ->
match restorables.TryGetValue nodeId with
| true, restorable -> Some restorable
| _ -> None)
|> List.ofSeq

let callback() =
notification.NodeDownloading node
match cache.TryGetSummary allowRemoteCache cacheEntryId with
| Some summary ->
Log.Debug("{NodeId} restoring '{Project}/{Target}' with {Hash}", node.Id, node.Project, node.Target, node.TargetHash)
match summary.Outputs with
| Some outputs ->
let files = IO.enumerateFiles outputs
IO.copyFiles projectDirectory outputs files |> ignore
api |> Option.iter (fun api -> api.UseArtifact node.ProjectHash node.TargetHash)
| _ -> ()
notification.NodeCompleted node TaskRequest.Restore true
| _ ->
notification.NodeCompleted node TaskRequest.Restore false
raiseBugError $"Unable to download build output for {cacheEntryId} for node {node.Id}"

if node.Managed then
let restorable = Restorable(callback, dependencies)
restorables.TryAdd(node.Id, restorable) |> ignore
else
Log.Debug("{NodeId} skipping restore unmanaged '{Project}/{Target}' with {Hash}", node.Id, node.Project, node.Target, node.TargetHash)
notification.NodeCompleted node TaskRequest.Restore true
if summary.IsSuccessful then TaskStatus.Success summary.EndedAt |> Some
else TaskStatus.Failure (summary.EndedAt, $"Restored node {node.Id} with a build in failure state") |> Some
| _ -> None
Log.Debug("{NodeId} is marked as used", node.Id)
TaskRequest.Restore, restoreNode()
| _ ->
Log.Debug("{NodeId} must be build since no summary and required", node.Id)
TaskRequest.Build, buildNode()
else
Log.Debug("{NodeId} is not cacheable", node.Id)
TaskRequest.Build, buildNode()


let hub = Hub.Create(options.MaxConcurrency)
let rec schedule nodeId =
if nodeResults.TryAdd(nodeId, (TaskRequest.Build, TaskStatus.Pending)) then
let node = graph.Nodes[nodeId]
let nodeComputed = hub.GetSignal<DateTime> nodeId
match completionStatus with
| Some completionStatus ->
Log.Debug("{NodeId} completed restore request with status {Status}", node.Id, completionStatus)
nodeResults[node.Id] <- (TaskRequest.Restore, completionStatus)
let success, completionDate =
match completionStatus with
| TaskStatus.Success completionDate -> true, completionDate
| TaskStatus.Failure (completionDate, _) -> false, completionDate
| _ -> raiseBugError "Unexpected pending state"
notification.NodeCompleted node TaskRequest.Restore success
if success then nodeComputed.Value <- completionDate
| _ ->
// await dependencies
let awaitedDependencies =
node.Dependencies |> Seq.map (fun awaitedProjectId ->
schedule awaitedProjectId
hub.GetSignal<DateTime> awaitedProjectId)
|> List.ofSeq

let onAllSignaled () =
try
let completionStatus = processNode node
Log.Debug("{NodeId} completed build request with status {Status}", node.Id, completionStatus)
nodeResults[node.Id] <- (TaskRequest.Build, completionStatus)
let success, completionDate =
match completionStatus with
| TaskStatus.Success completionDate -> true, completionDate
| TaskStatus.Failure (completionDate, _) -> false, completionDate
| _ -> raiseBugError "Unexpected pending state"
notification.NodeCompleted node TaskRequest.Build success
if success then nodeComputed.Value <- completionDate
with
| exn ->
Log.Fatal(exn, "{NodeId} failed on build request", node.Id)
// await dependencies
let awaitedDependencies =
node.Dependencies
|> Seq.map (fun awaitedProjectId ->
schedule awaitedProjectId
hub.GetSignal<DateTime> awaitedProjectId)
|> List.ofSeq

let onAllSignaled () =
try
let maxCompletionChildren =
match awaitedDependencies with
| [ ] -> DateTime.MinValue
| _ -> awaitedDependencies |> Seq.maxBy (fun dep -> dep.Value) |> (fun dep -> dep.Value)

let buildRequest, completionStatus = processNode maxCompletionChildren node
Log.Debug("{NodeId} completed request {Request} with status {Status}", node.Id, buildRequest, completionStatus)
let success, completionDate =
match completionStatus with
| TaskStatus.Success completionDate -> true, completionDate
| TaskStatus.Failure (completionDate, _) -> false, completionDate
| _ -> raiseBugError "Unexpected pending state"
nodeResults[node.Id] <- (buildRequest, completionStatus)
notification.NodeCompleted node buildRequest success
if success then nodeComputed.Value <- completionDate
with
exn ->
Log.Fatal(exn, "{NodeId} unexpected failure while building", node.Id)
nodeResults[node.Id] <- (TaskRequest.Build, TaskStatus.Failure (DateTime.UtcNow, exn.Message))
notification.NodeCompleted node TaskRequest.Build false
reraise()

let awaitedSignals = awaitedDependencies |> List.map (fun entry -> entry :> ISignal)
hub.Subscribe nodeId awaitedSignals onAllSignaled
let awaitedSignals = awaitedDependencies |> List.map (fun entry -> entry :> ISignal)
hub.Subscribe nodeId awaitedSignals onAllSignaled

graph.RootNodes |> Seq.iter schedule

Expand Down Expand Up @@ -402,12 +416,13 @@ let loadSummary (options: ConfigOptions.Options) (cache: Cache.ICache) (graph: G
let branchOrTag = options.BranchOrTag

let endedAt = DateTime.UtcNow
let buildInfo = { Summary.Commit = headCommit.Sha
Summary.BranchOrTag = branchOrTag
Summary.StartedAt = startedAt
Summary.EndedAt = endedAt
Summary.IsSuccess = isSuccess
Summary.Targets = options.Targets
Summary.Nodes = nodeStatus }
let buildInfo =
{ Summary.Commit = headCommit.Sha
Summary.BranchOrTag = branchOrTag
Summary.StartedAt = startedAt
Summary.EndedAt = endedAt
Summary.IsSuccess = isSuccess
Summary.Targets = options.Targets
Summary.Nodes = nodeStatus }
buildInfo

2 changes: 1 addition & 1 deletion src/Terrabuild/Core/Builder.fs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -118,7 +118,7 @@ let build (options: ConfigOptions.Options) (configuration: Configuration.Workspa

let managed = target.Managed |> Option.defaultValue true

let rebuild = target.Rebuild
let rebuild = options.Force || target.Rebuild

let targetOutput =
if managed then target.Outputs
Expand Down
14 changes: 7 additions & 7 deletions tests/cluster-layers/results/terrabuild-debug.build-graph.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -78,7 +78,7 @@
],
"cache": 3,
"managed": true,
"rebuild": false,
"rebuild": true,
"isLeaf": true
},
"b:build": {
Expand Down Expand Up @@ -159,7 +159,7 @@
],
"cache": 3,
"managed": true,
"rebuild": false,
"rebuild": true,
"isLeaf": true
},
"c:build": {
Expand Down Expand Up @@ -240,7 +240,7 @@
],
"cache": 3,
"managed": true,
"rebuild": false,
"rebuild": true,
"isLeaf": true
},
"d:build": {
Expand Down Expand Up @@ -301,7 +301,7 @@
],
"cache": 3,
"managed": true,
"rebuild": false,
"rebuild": true,
"isLeaf": true
},
"e:build": {
Expand Down Expand Up @@ -362,7 +362,7 @@
],
"cache": 3,
"managed": true,
"rebuild": false,
"rebuild": true,
"isLeaf": true
},
"f:build": {
Expand Down Expand Up @@ -421,7 +421,7 @@
],
"cache": 3,
"managed": true,
"rebuild": false,
"rebuild": true,
"isLeaf": true
},
"g:build": {
Expand Down Expand Up @@ -479,7 +479,7 @@
],
"cache": 3,
"managed": true,
"rebuild": false,
"rebuild": true,
"isLeaf": true
}
},
Expand Down
Loading