Partial (per-branch) dependencies for remote workflows via law.eager

[kandrosov]

Problem

A remote workflow is a single luigi task, and luigi never runs a task before all of its
requirements are complete. A workflow that requires another workflow therefore cannot submit a
single job until the required workflow has finished entirely — even for branches whose own
inputs landed hours earlier. LocalWorkflow does not have this problem, because its proxy yields
branch tasks as luigi dynamic requirements and luigi then resolves dependencies per branch.

Two facts from the source bound the solution space:

• luigi never dispatches a task while a declared dependency is incomplete (Scheduler._schedulable,
  plus a second check_unfulfilled_deps gate in the worker), with no configuration escape. So
  concurrency can only come from changing what is declared.
• There is no luigi-schedulable object for "one remote branch": a branch task always runs in the
  submitting process, and the workflow instance is the unit luigi schedules. That is why
  examples/sequential_htcondor_at_cern has to create one workflow instance per chunk via the
  significant branches parameter.

That workaround does not scale. Its cost grows with the number of chunks rather than the number of
stages, and it duplicates work whenever two downstream branches share an upstream branch. Measured
on the example added here (12 upstream branches feeding 6 downstream branches with overlapping
requirements):

	this PR	one instance per chunk
luigi tasks	2	13
poll loops / submission contexts	2	12
job-data control files	2	12
--workers for full concurrency	1	8
upstream branches executed more than once	0	5 (up to 3x)

What this adds

One marker, in one place. The per-branch mapping is read off the requires() the task already has,
so nothing is declared twice:

class Combine(law.htcondor.HTCondorWorkflow):

    def workflow_requires(self):
        reqs = super().workflow_requires()
        reqs["produce"] = law.eager(Produce.req(self, branch=-1))   # the only change
        return reqs

    def requires(self):
        # unchanged: these per-branch requirements are what the submission is gated on
        return {b: Produce.req(self, branch=b) for b in self.branch_data}

Measured with the mock batch system in the new example (Produce has 12 branches, branch 0 runs
25x longer than the rest; Combine has 4 branches needing 3 consecutive Produce branches each):
the three unblocked Combine jobs are submitted at t = 3.0 s instead of t = 14.5 s, and the fourth
right after the straggler lands. On real HTCondor at CERN with a 150 s straggler, Combine jobs
were submitted at t = 135 s and t = 151 s while the last Produce branch finished at t = 244 s.

How it works

1. law.eager(w) replaces the requirement with an EagerRequirement, a law.WrapperTask that
   requires everything w requires, but not w itself. luigi therefore still builds w's own
   inputs, bundles and credentials in the usual order, and — being a wrapper task — the marker is
   complete once those are, so the depending workflow starts right away. Nothing reports itself
   complete when it is not. output() is forwarded, so workflow_input() is unaffected, and
   requires() is forwarded, so --print-status, --print-deps and --remove-output traverse
   through it (it shows up in the tree with an eager flag).
2. The depending workflow's proxy drives w in a daemon thread (ThreadEagerDriver), so w
   submits and polls its jobs while the depending workflow is already working. Each stage keeps its
   own job data, job grouping, polling interval and backend, which is what lets a CRAB stage feed
   an HTCondor stage. No new luigi task is created — this matters, because luigi's own scheduling
   cost grows roughly quadratically with graph size (measured: 45 s at 4 000 tasks, 188 s at 8 000),
   so a luigi node per branch is not viable at production scale.
3. Submission is gated per job inside the machinery that already exists.
   BaseRemoteWorkflowProxy registers every branch chunk in job_data.unsubmitted_jobs and re-runs
   submit() on every poll iteration — that is how --parallel-jobs throttling works. A job whose
   requirements are not met yet is skipped without being popped, so len(job_data), and with it
   poll()'s n_jobs snapshot, never changes.
4. Readiness is decided from the output collection of each eager requirement — one bulk
   existence check per requirement per poll iteration rather than one per branch — and is
   accumulated monotonically, so a temporarily unreachable storage cannot revoke it. Job status is
   deliberately not used as the signal: a job can report FINISHED without its outputs existing.
5. Only one consumer drives a given requirement. Two workflows can require the same one eagerly, so
   the right to drive is claimed in a law.util.mp_manager dictionary shared across the processes of
   one law run, and the others observe its state through the same registry. Without the claim the
   requirement's jobs are submitted twice and the second driver crashes; the
   shared_requirement scenario covers it.
6. A job whose requirements can no longer appear — the driven requirement stopped running and they
   are still missing — is failed with a diagnosis naming them
   (missing branch(es) of eager requirement(s), produce: [5] (requirement failed with: ...))
   instead of hanging. It goes through the normal failed-job path, so tolerance and acceptance
   apply unchanged.

Workflows with many branches can state the mapping directly instead of letting law derive it from
requires(), which instantiates one task per dependency edge:

    def eager_branch_dependencies(self):
        # {own branch: {requirement key: upstream branches}}
        return {b: {"produce": self.branch_map[b]} for b in self.branch_map}

The difference is not cosmetic at scale: on a 7 600-branch consumer with 380 000 edges, deriving
takes about 1 590 s while the stated mapping takes 0.14 s.

What is unchanged

Everything that does not use the marker. submit() skips the gate entirely when no requirement is
eager, and BaseWorkflowProxy.requires() returns its input object unchanged when no marker is
present — the added cost there is one lazy structure scan, measured at 1.3 ms over a
2 000-object requirement structure. --workflow local is unaffected: the marker is unwrapped for
non-remote proxies, so luigi resolves branch dependencies as before (the local scenario asserts
that no EagerRequirement reaches the graph).

Three call sites that assumed luigi's worker-injected callbacks always exist are now guarded the way
publish_message and publish_progress already were: scheduler_messages, set_tracking_url and
decrease_running_resources are attached by TaskProcess.forward_reporter_attributes only while a
worker runs the task, and are removed afterwards. Any code that runs a proxy outside a worker hit
AttributeError before.

Polling status lines are prefixed with the task family when several stages poll at once, since they
otherwise interleave without saying which stage they belong to.

Limitations

• A branch map that does not exist yet cannot be gated per branch. If the depending workflow's
  branch map is produced by a dynamic_workflow_condition on the eager requirement, there are no
  branches to hold back, and submitting the placeholder branch would submit a job for nothing. That
  combination is refused with an explicit message rather than guessed at. Letting a branch map
  grow as upstream branches land is a separate, larger change.
• The depending workflow can complete while the requirement has not, if a branch of the
  requirement that nothing depends on failed. That is the intended semantics — a branch waits only
  for what it consumes — and the failure is logged as a warning.
• With no_poll there is no loop that could release held-back jobs, so nothing is driven and,
  unless everything is already in place, nothing is submitted.
• A backend that batches submission deliberately (CRAB creates one task per submission wave) needs
  its own aggregation policy on top. The readiness gate only filters the candidate set and
  poll_callback still decides when a wave goes out. That combination has not been exercised
  against a real CRAB task.

Validation

examples/partial_dependencies/ contains a HelloWorld payload and a mock batch system — real
subprocesses, real job ids, real polling, state on disk — so the code path through
law.workflow.remote is the one HTCondor, Slurm and CRAB take, without needing a batch system. A
full pass takes a few minutes:

source setup.sh
law index --verbose
python test.py

14 scenarios, all passing:

scenario	asserts
barrier	law's default really does wait for the whole upstream (the baseline being improved on)
eager	branches whose inputs exist are submitted while the straggler runs
overlap	non-contiguous, overlapping dependencies; no branch submitted twice
chain	three eager stages pipeline; a stage-3 branch finishes before stage 1 does
upstream_failure	a permanently failed upstream branch fails exactly the dependent job, with a diagnosis, and does not hang; independent branches still complete
blocked_not_retried	after a restart, a job whose input will never exist is not resubmitted into a slot that could only fail
batching	with tasks_per_job > 1 a job waits for the slowest branch it covers
upstream_complete	an already complete requirement behaves exactly as before: one submission burst, nothing driven
local	--workflow local is unchanged
resume	a killed run resumes without resubmitting or re-running anything
shared_requirement	two consumers of one eager requirement drive it once, not twice
driven_requirements	the driven workflow's own requirements are built by luigi first, so its jobs never build them on a worker
dynamic_branch_map	the unsupported combination is refused, with no job submitted
chunked_workaround	quantifies the cost and the duplicate work of the existing workaround

The same payload was run end to end on real HTCondor at CERN (--workflow htcondor), and
--print-status -1 and --print-deps were checked by hand. flake8 is clean on the changed files.

Three defects in earlier revisions of this branch were each found and then confirmed by reverting
the fix and watching the scenario fail: the wrapper originally faked complete() and so skipped the
driven workflow's own subtree; a restart resubmitted permanently-blocked jobs through the retry path;
and the per-branch derivation matched requirements by task_id, which fails open — returning an
empty dependency map, read as "every branch is ready" — when a project refers to the same workflow
with a different branch selection. Matching now ignores the branch-selecting parameters, and
branches that still cannot be mapped are gated on the real complete() of their requirements.

Files

file	what
law/workflow/eager.py	new: the marker, the identity matching and the thread driver
law/workflow/remote.py	the readiness gate in submit(), the eager state on the proxy, the guards
law/workflow/base.py	unwrap markers in BaseWorkflowProxy.requires(); guard scheduler_messages
law/__init__.py	export law.eager
docs/api/workflow/eager.rst	API page, added to the workflow toctree
examples/partial_dependencies/	the example, the mock batch system and the scenarios