A technical PM case study about prerequisites, blockers, rework, and launch order.
This project models work as tasks with dependencies. The code can identify which tasks are ready, prevent tasks from being completed before prerequisites are done, run a schedule step by step, and handle redo logic when finished work becomes invalid.
How can a program decide what work can happen now, what work is blocked, and what has to be redone when a dependency changes? This is close to a product manager's planning problem: launches depend on prerequisites, some work can happen in parallel, and changes can force downstream work to be revisited.
Click through the demo to see the dependency logic working. The scheduler only moves a task into Ready when its prerequisites are complete.
Ready to start. The first unblocked task is available.
| Code Element | What It Represents | PM Meaning |
|---|---|---|
| tasks | Every task known to the scheduler. | The work items in a launch plan or roadmap. |
| predecessors | Tasks that must be completed before another task. | Prerequisites and blockers. |
| successors | Tasks that depend on a completed task. | Downstream work affected by a decision. |
| completed_tasks | Tasks already marked as done. | Current project status. |
from collections import defaultdict
class DependencyScheduler(object):
def __init__(self):
self.tasks = set()
self.successors = defaultdict(set)
self.predecessors = defaultdict(set)
self.completed_tasks = set()
def add_task(self, t, dependencies):
assert t not in self.tasks or len(self.predecessors[t]) == 0, "The task was already present."
self.tasks.add(t)
self.tasks.update(dependencies)
self.predecessors[t] = set(dependencies)
for u in dependencies:
self.successors[u].add(t)
def reset(self):
self.completed_tasks = set()
@property
def done(self):
return self.completed_tasks == self.tasks
@property
def uncompleted(self):
return self.tasks - self.completed_tasksThe scheduler stores each task, what it depends on, what depends on it, and what has already been completed. This is the foundation for turning a task list into a dependency-aware plan.
def scheduler_available_tasks(self):
available = set()
for task in self.tasks:
if task not in self.completed_tasks:
if self.predecessors[task].issubset(self.completed_tasks):
available.add(task)
return available
def scheduler_mark_completed(self, t):
self.completed_tasks.add(t)
newly_available = set()
for task in self.successors[t]:
if task not in self.completed_tasks:
if self.predecessors[task].issubset(self.completed_tasks):
newly_available.add(task)
return newly_available
DependencyScheduler.available_tasks = property(scheduler_available_tasks)
DependencyScheduler.mark_completed = scheduler_mark_completedA task becomes available only when all of its prerequisites are complete. In PM terms, this separates work that can start now from work that is still blocked.
class RunSchedule(object):
def __init__(self, scheduler):
self.scheduler = scheduler
self.in_process = None
def reset(self):
self.scheduler.reset()
self.in_process = None
def step(self):
if self.in_process is None:
self.in_process = self.scheduler.available_tasks
if len(self.in_process) == 0:
return None
t = random.choice(list(self.in_process))
self.in_process = self.in_process - {t} | self.scheduler.mark_completed(t)
return t
@property
def done(self):
return self.scheduler.done
def run(self):
tasks = []
while not self.done:
t = self.step()
if t is not None:
tasks.append(t)
return tasksThe runner turns available-task logic into a working schedule. Each step completes one available task and unlocks any newly available work.
class IllegalCompletion(Exception):
def __init__(self, t):
self.message = "Not all prerequisited for %r have been completed" % t
def dependency_scheduler_mark_completed(self, t):
if not self.predecessors[t].issubset(self.completed_tasks):
raise IllegalCompletion(t)
self.completed_tasks.add(t)
return {u for u in self.successors[t]
if self.predecessors[u].issubset(self.completed_tasks)}
DependencyScheduler.mark_completed = dependency_scheduler_mark_completedThis adds a safeguard: a task cannot be marked done if its prerequisites are unfinished. That is a useful product-planning concept because it prevents a plan from pretending blocked work is complete.
def dependency_scheduler_redo(self, t):
to_redo = set()
stack = [t]
while stack:
current = stack.pop()
if current not in to_redo:
to_redo.add(current)
stack.extend(self.successors[current])
self.completed_tasks -= to_redo
return to_redo
DependencyScheduler.redo = dependency_scheduler_redoIf one task needs to be redone, all downstream tasks that depend on it may also need to be redone. This models a common launch-planning issue: changing a requirement can invalidate later work.
def dependency_scheduler_cooking_redo(self, v):
to_redo = set()
to_redo.add(v)
forward_tasks = set([v])
while forward_tasks:
current = forward_tasks.pop()
for successor in self.successors[current]:
if successor not in to_redo and successor in self.completed_tasks:
to_redo.add(successor)
forward_tasks.add(successor)
backward_tasks = set(to_redo.copy())
while backward_tasks:
current = backward_tasks.pop()
for predecessor in self.predecessors[current]:
if predecessor not in to_redo and predecessor in self.completed_tasks:
to_redo.add(predecessor)
backward_tasks.add(predecessor)
self.completed_tasks -= to_redo
return to_redo
DependencyScheduler.cooking_redo = dependency_scheduler_cooking_redoThis version handles a stricter rework case by propagating redo both forward and backward. It shows deeper dependency reasoning because completed prerequisites can also become invalid when the central task changes.
class AND_OR_Scheduler(object):
def __init__(self):
self.tasks = set()
self.predecessors = defaultdict(set)
self.successors = defaultdict(set)
self.completed_tasks = set()
self.task_type = {}
def add_and_task(self, t, dependencies):
self.tasks.add(t)
self.tasks.update(dependencies)
self.predecessors[t] = set(dependencies)
self.task_type[t] = "AND"
for u in dependencies:
self.successors[u].add(t)
if u not in self.task_type:
self.task_type[u] = "AND"
def add_or_task(self, t, dependencies):
self.tasks.add(t)
self.tasks.update(dependencies)
self.predecessors[t] = set(dependencies)
self.task_type[t] = "OR"
for u in dependencies:
self.successors[u].add(t)
if u not in self.task_type:
self.task_type[u] = "AND"
@property
def available_tasks(self):
available = set()
for task in self.tasks:
if task in self.completed_tasks:
continue
if self.task_type[task] == "AND":
if self.predecessors[task].issubset(self.completed_tasks):
available.add(task)
else:
if self.predecessors[task] and not self.predecessors[task].isdisjoint(self.completed_tasks):
available.add(task)
elif not self.predecessors[task]:
available.add(task)
return availableAND tasks require every prerequisite. OR tasks become available when at least one prerequisite is complete. That maps nicely to product planning when one feature has mandatory prerequisites while another can be unlocked by multiple possible paths.
| Feature | Based On Code | PM Value |
|---|---|---|
| Ready Now | available_tasks | Shows what the team can start immediately. |
| Blocked Work | predecessors not completed | Makes blockers visible before they delay a launch. |
| Suggested Sequence | RunSchedule.run() | Turns a dependency map into an execution order. |
| Redo Impact | redo and cooking_redo | Shows what downstream work is affected by a change. |
| Alternative Paths | AND_OR_Scheduler | Models cases where one of several prerequisites can unblock work. |