377 lines
13 KiB
Python
377 lines
13 KiB
Python
"""
|
|
CWE-407: Algorithmic Complexity - Luigi scheduler._upstream_status
|
|
|
|
luigi-0001: _upstream_status missing in_stack guard — O(edges) vs O(nodes)
|
|
|
|
In luigi/scheduler.py lines 1300-1303:
|
|
|
|
if dep_id not in upstream_status_table:
|
|
if dep.status == PENDING and dep.deps:
|
|
task_stack += [dep_id] + list(dep.deps) # BUG: no guard
|
|
upstream_status_table[dep_id] = ""
|
|
|
|
When multiple PENDING tasks share downstream dependencies (diamond/DAG
|
|
pattern), each parent unconditionally pushes all its children onto
|
|
task_stack — including children that are already queued or already resolved.
|
|
This means every edge in the dependency graph generates a stack entry,
|
|
giving O(E) total stack operations where E = number of edges.
|
|
|
|
In a dense DAG (e.g. complete DAG where node i depends on all nodes j>i),
|
|
E = O(N^2), so the algorithm is O(N^2) while the fixed version is O(N).
|
|
|
|
The fix adds an in_stack set so each child is enqueued at most once,
|
|
bounding stack operations to O(N) regardless of edge density.
|
|
|
|
Measured ratio at N=30: ~8x (defect=527 pops, fixed=63 pops).
|
|
Measured ratio at N=50: ~13x (defect=1377 pops, fixed=103 pops).
|
|
"""
|
|
|
|
import sys
|
|
|
|
PENDING = "PENDING"
|
|
DONE = "DONE"
|
|
FAILED = "FAILED"
|
|
|
|
STATUS_TO_UPSTREAM_MAP = {
|
|
FAILED: "UPSTREAM_FAILED",
|
|
PENDING: "UPSTREAM_PENDING",
|
|
}
|
|
UPSTREAM_SEVERITY_ORDER = [
|
|
"", "UPSTREAM_PENDING", "UPSTREAM_DISABLED",
|
|
"UPSTREAM_FAILED", "UPSTREAM_MISSING_INPUT",
|
|
]
|
|
|
|
|
|
def UPSTREAM_SEVERITY_KEY(s):
|
|
try:
|
|
return UPSTREAM_SEVERITY_ORDER.index(s)
|
|
except ValueError:
|
|
return -1
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Minimal task/state stubs (mirror luigi internals just enough for simulation)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class FakeTask:
|
|
def __init__(self, task_id, status, deps):
|
|
self.id = task_id
|
|
self.status = status
|
|
self.deps = list(deps)
|
|
|
|
|
|
class FakeState:
|
|
def __init__(self, tasks):
|
|
self._tasks = {t.id: t for t in tasks}
|
|
|
|
def has_task(self, task_id):
|
|
return task_id in self._tasks
|
|
|
|
def get_task(self, task_id):
|
|
return self._tasks.get(task_id)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Faithful simulation of the defective implementation
|
|
# (mirrors scheduler.py lines 1288-1312 exactly, injecting a pop counter)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def upstream_status_defect(task_id, state):
|
|
"""
|
|
Defective _upstream_status: no in_stack guard.
|
|
Returns (upstream_status, pop_count) where pop_count measures total work.
|
|
"""
|
|
upstream_status_table = {}
|
|
if not state.has_task(task_id):
|
|
return ("", 0)
|
|
|
|
task_stack = [task_id]
|
|
pop_count = 0
|
|
|
|
while task_stack:
|
|
dep_id = task_stack.pop()
|
|
pop_count += 1
|
|
dep = state.get_task(dep_id)
|
|
if dep:
|
|
if dep.status == DONE:
|
|
continue
|
|
if dep_id not in upstream_status_table:
|
|
if dep.status == PENDING and dep.deps:
|
|
# BUG: pushes all deps unconditionally — duplicates accumulate
|
|
task_stack += [dep_id] + list(dep.deps)
|
|
upstream_status_table[dep_id] = ""
|
|
else:
|
|
dep_status = STATUS_TO_UPSTREAM_MAP.get(dep.status, "")
|
|
upstream_status_table[dep_id] = dep_status
|
|
elif upstream_status_table[dep_id] == "" and dep.deps:
|
|
status = max(
|
|
(upstream_status_table.get(a, "") for a in dep.deps),
|
|
key=UPSTREAM_SEVERITY_KEY,
|
|
)
|
|
upstream_status_table[dep_id] = status
|
|
|
|
return (upstream_status_table.get(dep_id, ""), pop_count)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Fixed implementation
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def upstream_status_fixed(task_id, state):
|
|
"""
|
|
Fixed _upstream_status: in_stack guard prevents duplicate enqueue.
|
|
Returns (upstream_status, pop_count).
|
|
"""
|
|
upstream_status_table = {}
|
|
if not state.has_task(task_id):
|
|
return ("", 0)
|
|
|
|
task_stack = [task_id]
|
|
in_stack = {task_id}
|
|
pop_count = 0
|
|
|
|
while task_stack:
|
|
dep_id = task_stack.pop()
|
|
in_stack.discard(dep_id)
|
|
pop_count += 1
|
|
dep = state.get_task(dep_id)
|
|
if dep:
|
|
if dep.status == DONE:
|
|
continue
|
|
if dep_id not in upstream_status_table:
|
|
if dep.status == PENDING and dep.deps:
|
|
task_stack.append(dep_id)
|
|
in_stack.add(dep_id)
|
|
upstream_status_table[dep_id] = ""
|
|
for child_id in dep.deps:
|
|
if child_id not in upstream_status_table and child_id not in in_stack:
|
|
task_stack.append(child_id)
|
|
in_stack.add(child_id)
|
|
else:
|
|
dep_status = STATUS_TO_UPSTREAM_MAP.get(dep.status, "")
|
|
upstream_status_table[dep_id] = dep_status
|
|
elif upstream_status_table[dep_id] == "" and dep.deps:
|
|
status = max(
|
|
(upstream_status_table.get(a, "") for a in dep.deps),
|
|
key=UPSTREAM_SEVERITY_KEY,
|
|
)
|
|
upstream_status_table[dep_id] = status
|
|
|
|
return (upstream_status_table.get(dep_id, ""), pop_count)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Graph constructors
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def make_complete_dag(n):
|
|
"""
|
|
Complete DAG of depth n + leaf.
|
|
|
|
- root (PENDING) depends on [n0, n1, ..., n_{n-1}]
|
|
- node n_i (PENDING) depends on [n_{i+1}, ..., n_{n-1}, leaf]
|
|
- leaf (FAILED) has no deps
|
|
|
|
Edge count E = O(N^2): node i has (N-i) outgoing edges.
|
|
The defect processes O(E) = O(N^2) stack entries.
|
|
The fix processes O(N) stack entries.
|
|
"""
|
|
tasks = []
|
|
node_ids = [f"n{i}" for i in range(n)]
|
|
tasks.append(FakeTask("root", PENDING, node_ids))
|
|
for i in range(n):
|
|
forward_deps = [f"n{j}" for j in range(i + 1, n)] + ["leaf"]
|
|
tasks.append(FakeTask(f"n{i}", PENDING, forward_deps))
|
|
tasks.append(FakeTask("leaf", FAILED, []))
|
|
return FakeState(tasks)
|
|
|
|
|
|
def make_diamond_state(n_parents):
|
|
"""
|
|
Simple diamond: root (PENDING) -> [p0..p_{N-1}] (PENDING) -> shared (FAILED).
|
|
Each parent unconditionally pushes shared; the fix enqueues it once.
|
|
"""
|
|
tasks = []
|
|
parent_ids = [f"p{i}" for i in range(n_parents)]
|
|
tasks.append(FakeTask("root", PENDING, parent_ids))
|
|
for pid in parent_ids:
|
|
tasks.append(FakeTask(pid, PENDING, ["shared"]))
|
|
tasks.append(FakeTask("shared", FAILED, []))
|
|
return FakeState(tasks)
|
|
|
|
|
|
def make_linear_chain():
|
|
"""A -> B -> C -> D(FAILED): no diamonds, both versions behave identically."""
|
|
return FakeState([
|
|
FakeTask("A", PENDING, ["B"]),
|
|
FakeTask("B", PENDING, ["C"]),
|
|
FakeTask("C", PENDING, ["D"]),
|
|
FakeTask("D", FAILED, []),
|
|
])
|
|
|
|
|
|
def make_all_done():
|
|
"""root -> [a, b] all DONE: upstream status should be ''."""
|
|
return FakeState([
|
|
FakeTask("root", PENDING, ["a", "b"]),
|
|
FakeTask("a", DONE, []),
|
|
FakeTask("b", DONE, []),
|
|
])
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_complete_dag_complexity():
|
|
"""
|
|
Complete DAG at N=30: O(N^2) edges → defect does ~8x more pops than fix.
|
|
This is the canonical demonstration of the O(edges) vs O(nodes) gap.
|
|
"""
|
|
n = 30
|
|
state = make_complete_dag(n)
|
|
|
|
_, defect_pops = upstream_status_defect("root", state)
|
|
_, fixed_pops = upstream_status_fixed("root", state)
|
|
|
|
ratio = defect_pops / fixed_pops
|
|
assert ratio > 5, (
|
|
f"test_complete_dag_complexity: expected ratio > 5x, got {ratio:.1f}x "
|
|
f"(defect={defect_pops} pops, fixed={fixed_pops} pops)"
|
|
)
|
|
print(
|
|
f"PASS test_complete_dag_complexity: "
|
|
f"defect={defect_pops} pops, fixed={fixed_pops} pops, ratio={ratio:.1f}x"
|
|
)
|
|
|
|
|
|
def test_complete_dag_scaling():
|
|
"""
|
|
Ratio must grow with N, confirming super-linear defect growth.
|
|
Ratio at N=50 should exceed ratio at N=20 by a comfortable margin.
|
|
"""
|
|
def measure(n):
|
|
s = make_complete_dag(n)
|
|
_, d = upstream_status_defect("root", s)
|
|
_, f = upstream_status_fixed("root", s)
|
|
return d / f
|
|
|
|
ratio_20 = measure(20)
|
|
ratio_50 = measure(50)
|
|
|
|
assert ratio_50 > ratio_20 * 1.5, (
|
|
f"test_complete_dag_scaling: ratio should grow with N; "
|
|
f"ratio@20={ratio_20:.1f}x ratio@50={ratio_50:.1f}x"
|
|
)
|
|
print(
|
|
f"PASS test_complete_dag_scaling: "
|
|
f"ratio@N=20={ratio_20:.1f}x → ratio@N=50={ratio_50:.1f}x (growing)"
|
|
)
|
|
|
|
|
|
def test_correctness_complete_dag():
|
|
"""
|
|
Both implementations must return the same upstream status for the complete DAG.
|
|
The leaf is FAILED, so root should report UPSTREAM_FAILED.
|
|
"""
|
|
for n in [5, 15, 30]:
|
|
state = make_complete_dag(n)
|
|
defect_result, _ = upstream_status_defect("root", state)
|
|
fixed_result, _ = upstream_status_fixed("root", state)
|
|
assert defect_result == fixed_result, (
|
|
f"test_correctness_complete_dag N={n}: "
|
|
f"defect='{defect_result}', fixed='{fixed_result}'"
|
|
)
|
|
assert fixed_result == "UPSTREAM_FAILED", (
|
|
f"test_correctness_complete_dag N={n}: expected UPSTREAM_FAILED, got '{fixed_result}'"
|
|
)
|
|
print("PASS test_correctness_complete_dag: status=UPSTREAM_FAILED for N=5,15,30")
|
|
|
|
|
|
def test_diamond_shared_node():
|
|
"""
|
|
Simple diamond: N parents all depending on one shared leaf.
|
|
Defect pushes shared N times; fix pushes it once.
|
|
"""
|
|
n = 20
|
|
state = make_diamond_state(n)
|
|
|
|
_, defect_pops = upstream_status_defect("root", state)
|
|
_, fixed_pops = upstream_status_fixed("root", state)
|
|
|
|
# defect pops shared_leaf N times wastefully; fixed pops it once
|
|
assert defect_pops > fixed_pops, (
|
|
f"test_diamond_shared_node: expected defect > fixed, "
|
|
f"got defect={defect_pops}, fixed={fixed_pops}"
|
|
)
|
|
print(
|
|
f"PASS test_diamond_shared_node: "
|
|
f"defect={defect_pops} pops, fixed={fixed_pops} pops "
|
|
f"(defect enqueues shared {n} times)"
|
|
)
|
|
|
|
|
|
def test_correctness_diamond():
|
|
"""Both versions must agree on upstream status for diamond graph."""
|
|
for n in [5, 20, 50]:
|
|
state = make_diamond_state(n)
|
|
defect_result, _ = upstream_status_defect("root", state)
|
|
fixed_result, _ = upstream_status_fixed("root", state)
|
|
assert defect_result == fixed_result, (
|
|
f"test_correctness_diamond N={n}: "
|
|
f"defect='{defect_result}', fixed='{fixed_result}'"
|
|
)
|
|
print("PASS test_correctness_diamond: status matches for N=5,20,50")
|
|
|
|
|
|
def test_no_regression_linear_chain():
|
|
"""
|
|
Linear chain has no shared nodes so both implementations do identical work.
|
|
Pop counts should be equal (no wasted pushes possible).
|
|
"""
|
|
state = make_linear_chain()
|
|
|
|
_, defect_pops = upstream_status_defect("A", state)
|
|
_, fixed_pops = upstream_status_fixed("A", state)
|
|
|
|
assert defect_pops == fixed_pops, (
|
|
f"test_no_regression_linear_chain: expected equal pops, "
|
|
f"got defect={defect_pops}, fixed={fixed_pops}"
|
|
)
|
|
print(
|
|
f"PASS test_no_regression_linear_chain: "
|
|
f"both={defect_pops} pops (linear chain, no difference)"
|
|
)
|
|
|
|
|
|
def test_done_tasks_are_skipped():
|
|
"""
|
|
Tasks with DONE status are skipped; upstream status of root should be ''.
|
|
Both versions must agree and return ''.
|
|
"""
|
|
state = make_all_done()
|
|
|
|
defect_result, _ = upstream_status_defect("root", state)
|
|
fixed_result, _ = upstream_status_fixed("root", state)
|
|
|
|
assert defect_result == fixed_result == "", (
|
|
f"test_done_tasks_are_skipped: "
|
|
f"expected '' got defect='{defect_result}', fixed='{fixed_result}'"
|
|
)
|
|
print("PASS test_done_tasks_are_skipped: both return '' when all deps DONE")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Entry point
|
|
# ---------------------------------------------------------------------------
|
|
|
|
if __name__ == "__main__":
|
|
test_complete_dag_complexity()
|
|
test_complete_dag_scaling()
|
|
test_correctness_complete_dag()
|
|
test_diamond_shared_node()
|
|
test_correctness_diamond()
|
|
test_no_regression_linear_chain()
|
|
test_done_tasks_are_skipped()
|
|
print("ALL PASS")
|
|
sys.exit(0)
|