1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
//! Per-service start scheduling for `up`/`create`.
//!
//! Split from `mod.rs` to keep that file within the source line limit. The
//! scheduler is a subject of its own: it is where the difference between
//! "levels" and "a dependency graph" lives, and the two deadlock hazards it
//! avoids are documented inline rather than being folded into `run_up`.
use std::collections::{HashMap, HashSet};
use crate::compose::types::ComposeFile;
use crate::error::Result;
use super::readiness::SharedReady;
use super::Engine;
impl Engine {
/// Start every scheduled service, each as soon as its own dependencies are
/// up. `levels` is used only for its membership and validation, not as a
/// barrier.
#[allow(clippy::too_many_arguments)]
pub(super) async fn start_services_by_dependency(
&self,
levels: &[Vec<String>],
file: &ComposeFile,
enabled: &HashSet<String>,
target_set: &Option<HashSet<String>>,
present: &HashSet<String>,
existing_hash: &HashMap<String, String>,
no_recreate: bool,
force_recreate: bool,
start: bool,
readiness: &HashMap<String, SharedReady<'_>>,
) -> Result<()> {
// Per-service scheduling, not level barriers. A service starts as soon
// as *its own* dependencies are up, rather than waiting for every
// service in the level ahead of it.
//
// The levels are still resolved, because that is what validates the
// graph (a cycle or a missing required dependency is an error before
// anything is created) — they just no longer gate execution.
//
// Measured cost of the barrier on a 41-service level with one dependent
// behind one of them: the dependent started 809ms after its dependency
// under barriers and 129ms under docker compose's per-service DAG,
// because a level is created with bounded concurrency and the barrier
// made the dependent wait for the last batch rather than for what it
// actually needed.
//
// Two things make this safe to do with plain futures:
//
// * `try_join_all` polls *every* future, so a service blocked on a
// dependency yields and the others make progress. `buffer_unordered`
// would deadlock here — it only polls the first N, and those N can
// all be waiting on a dependency that is queued behind them.
// * the concurrency bound is a semaphore held only across the actual
// work, never across a dependency wait, for the same reason.
//
// A failure cancels the rest: `try_join_all` drops the other futures on
// the first error, so a service still waiting on a dependency that will
// now never come up is dropped rather than hanging.
let scheduled: Vec<&str> = levels.iter().flatten().map(String::as_str).collect();
let permits = std::sync::Arc::new(tokio::sync::Semaphore::new(
super::parallel::MAX_LIFECYCLE_CONCURRENCY,
));
let done: std::collections::HashMap<&str, tokio::sync::watch::Sender<bool>> = scheduled
.iter()
.map(|name| (*name, tokio::sync::watch::channel(false).0))
.collect();
// Borrow the shared inputs once so the closure captures references
// rather than trying to move them per service.
let enabled = &enabled;
let target_set = &target_set;
let present = &present;
let existing_hash = &existing_hash;
let readiness = &readiness;
let started = scheduled.iter().map(|name| {
let permits = permits.clone();
let done = &done;
// Only the dependencies this service actually declares, and only
// those that are in this run's scheduled set — a service excluded by
// profiles or an explicit target list is not going to signal.
let deps: Vec<tokio::sync::watch::Receiver<bool>> = file
.services
.get(*name)
.map(|s| s.depends_on.service_names())
.unwrap_or_default()
.into_iter()
.filter_map(|d| done.get(d.as_str()).map(|tx| tx.subscribe()))
.collect();
async move {
for mut rx in deps {
// `changed()` errors only when every sender is gone, which
// here means the run is being torn down; treat it as done and
// let `try_join_all` surface the real error.
while !*rx.borrow_and_update() {
if rx.changed().await.is_err() {
break;
}
}
}
// The permit is taken *after* the waits, so a blocked service
// never occupies one.
let _permit = permits.acquire().await;
let result = self
.up_one_service(
name,
file,
enabled,
target_set,
present,
existing_hash,
no_recreate,
force_recreate,
start,
readiness,
)
.await;
if result.is_ok() {
if let Some(tx) = done.get(*name) {
let _ = tx.send(true);
}
}
result
}
});
futures_util::future::try_join_all(started).await?;
Ok(())
}
}