510 / 547 lines covered (93.2%)
1unit module BDD::Behave::Parallel;23use BDD::Behave::Parallel::Distribution;4use BDD::Behave::Parallel::WorkerPool;5use BDD::Behave::Parallel::Queue;6use BDD::Behave::Parallel::Manifest;7use BDD::Behave::Parallel::EventStream;8use BDD::Behave::SpecRegistry;9use BDD::Behave::SpecLoader;10use BDD::Behave::Runner;11use BDD::Behave::Formatter;12use BDD::Behave::Failures;13use BDD::Behave::Failure;14use BDD::Behave::Colors;15use BDD::Behave::Coverage;16use BDD::Behave::Configuration;1718need BDD::Behave::SpecTree;1920constant Suite = BDD::Behave::SpecTree::Suite;21constant ExampleGroup = BDD::Behave::SpecTree::ExampleGroup;22constant Example = BDD::Behave::SpecTree::Example;23constant RunResult = BDD::Behave::Runner::RunResult;2425class ParallelRunOptions is export {26 has Int $.worker-count is required;27 has @.spec-files;28 has @.worker-argv;29 has @.discovery-argv;30 has %.base-env;31 has BDD::Behave::Formatter $.formatter is required;32 has Bool $.verbose = False;33 has Int $.seed;34 has Str $.order = 'random';35 has Str $.seed-mode = 'xor';36 has Bool $.progress-total = False;37 has @.include-tags;38 has @.exclude-tags;39 has @.example-patterns;40 has @.only-locations;41 has Bool $.fail-fast-any = False;42 has Bool $.discovery-in-process = False;43 has Bool $.precompile-pass = True;44 has Str $.parallel-mode = BDD::Behave::Configuration::DEFAULT-PARALLEL-MODE;45 has Int $.parallel-retry = 0;46 has IO::Path $.coverage-log-dir = IO::Path;47 has Bool $.coverage-counts = False;48}4950sub discover-suites(@spec-files --> List) is export {51 my $registry = BDD::Behave::SpecRegistry::registry();52 my @suites;53 my @load-errors;54 for @spec-files -> $file {55 try {56 BDD::Behave::SpecLoader::load-spec-file($file);57 CATCH {58 default {59 @load-errors.push: %( :$file, :message(.message) );60 }61 }62 }63 my $suite = $registry.suite-for-file($file.IO);64 @suites.push($suite) if $suite.defined;65 }66 (@suites.List, @load-errors.List).List;67}6869sub current-include-flags(--> List) is export {70 (gather for $*REPO.repo-chain.reverse {71 take '-I' ~ .prefix.absolute if $_ ~~ CompUnit::Repository::FileSystem;72 }).List;73}7475sub default-discovery-argv(--> List) is export {76 ('raku', |current-include-flags(), $*PROGRAM-NAME).List;77}7879# Reaching the parallel runner costs a precompile subprocess, a discovery80# subprocess, and at least one worker subprocess, each paying the Raku81# startup and the module load again. For a selection that resolves to a82# single spec file that is more than the file itself costs, so an omitted83# --parallel resolves to zero workers there and the run stays in this84# process. An explicit --parallel N is always honored.85our sub resolve-worker-count(86 Int :$requested,87 Int :$file-count!,88 Int :$cpu-cores!,89 --> Int90) is export {91 return $requested if $requested.defined;92 return 0 if $file-count <= 1;93 $cpu-cores;94}9596# The pass exists to stop several processes writing the same precomp units from97# different load states, so it only pays for its extra Raku process when more98# than one process will load the spec files concurrently. A selection of one99# spec file is loaded by discovery and then by its single worker, one after the100# other, so the second process reads what the first wrote and there is nothing101# to protect against. Discovery in-process loads everything in this process for102# the same reason.103our sub should-precompile(104 Int :$file-count!,105 Bool :$requested = True,106 Bool :$discovery-in-process = False,107 --> Bool108) is export {109 return False unless $requested;110 return False if $discovery-in-process;111 $file-count > 1;112}113114# Build the module precompilation for every spec file in ONE --compile-only115# subprocess before discovery and the workers load anything, so the whole116# dependency graph is written by a single process in a single pass and every117# later process only reads it. Precomp units written by processes with118# different load states carry conflicting repossession records; loading such119# a mix segfaults MoarVM's deserializer, or surfaces as a deferred compile120# exception that crashes the exception printer at process teardown.121#122# An already-inconsistent cache (left by earlier mixed-writer runs) kills the123# pass itself the same way. The subprocess reports each project lib dir it124# loads through (via BEHAVE_COMPILE_REPORT); when the pass dies by signal,125# those dirs' .precomp caches are cleared and the pass runs once more to126# rebuild them consistently. Nonzero exits are ignored: discovery reports127# real load errors with their messages.128sub precompile-specs-serially(129 @spec-files,130 :@discovery-argv,131 :%base-env,132 --> Nil133) is export {134 return unless @spec-files.elems;135136 my @base = (@discovery-argv.elems ?? @discovery-argv.list !! default-discovery-argv()).list;137138 my $report = $*TMPDIR.add("behave-compile-report-{$*PID}-{(now * 1e6).Int}");139 LEAVE { try $report.unlink }140141 my %env = %base-env.elems ?? %base-env.Hash !! %*ENV.Hash;142 %env<BEHAVE_DISABLE_CONFIG> = '1';143 %env<BEHAVE_WORKER_INDEX> //= '0';144 %env<BEHAVE_WORKER_COUNT> //= '1';145 %env<BEHAVE_COMPILE_REPORT> = $report.absolute;146 %env<MVM_COVERAGE_LOG>:delete;147 %env<MVM_COVERAGE_CONTROL>:delete;148 %env<BEHAVE_COVERAGE_LOG>:delete;149150 my $timeout-secs = (%*ENV<BEHAVE_DISCOVERY_TIMEOUT> // 300).Numeric;151152 sub run-pass(@files --> Bool) {153 my @argv = |@base, '--no-config', '--compile-only', |@files.map(*.Str);154 my $proc = Proc::Async.new(|@argv);155 $proc.stdout.tap(-> $ { });156 $proc.stderr.tap(-> $ { });157158 my $died-by-signal = False;159 try {160 my $promise = $proc.start(:ENV(%env));161 await Promise.anyof($promise, Promise.in($timeout-secs));162 if $promise.status ~~ Kept {163 $died-by-signal = $promise.result.signal != 0;164 } else {165 $proc.kill(SIGKILL);166 try await $promise;167 $died-by-signal = True;168 }169 CATCH { default { $died-by-signal = True } }170 }171 $died-by-signal;172 }173174 return unless run-pass(@spec-files);175176 # The union pass crashed on an inconsistent cache. Loading one file per177 # process still works in that state, so run each file alone purely to learn178 # the project lib dirs the specs pull in (each process appends its repo179 # chain to the report), then clear those dirs' caches and rebuild the whole180 # graph in one pass.181 run-pass(($_,)) for @spec-files;182183 my @prefixes = $report.e ?? $report.slurp.lines.unique.grep(*.chars) !! ();184 return unless @prefixes.elems;185186 for @prefixes -> $prefix {187 my $precomp = $prefix.IO.add('.precomp');188 next unless $precomp.d;189 remove-tree($precomp);190 }191192 run-pass(@spec-files);193}194195# Recursive directory removal for the .precomp caches cleared above.196sub remove-tree(IO::Path:D $dir --> Nil) {197 for $dir.dir -> $entry {198 $entry.d ?? remove-tree($entry) !! (try $entry.unlink);199 }200 try $dir.rmdir;201}202203sub discover-suites-subprocess(204 @spec-files,205 :@discovery-argv,206 :%base-env,207 :$timeout,208 --> List209) is export {210 my @suites;211 my @load-errors;212213 return (@suites.List, @load-errors.List).List unless @spec-files.elems;214215 my @argv = (@discovery-argv.elems ?? @discovery-argv.list !! default-discovery-argv()).list;216 @argv.push: '--no-config';217 @argv.push: '--list-examples';218 @argv.push: '--list-examples-format=json';219 @argv.append: @spec-files.map(*.Str);220221 my %env = %base-env.elems ?? %base-env.Hash !! %*ENV.Hash;222 %env<BEHAVE_DISABLE_CONFIG> = '1';223 %env<BEHAVE_WORKER_INDEX> //= '0';224 %env<BEHAVE_WORKER_COUNT> //= '1';225 %env<MVM_COVERAGE_LOG>:delete;226 %env<MVM_COVERAGE_CONTROL>:delete;227 %env<BEHAVE_COVERAGE_LOG>:delete;228229 my $proc = Proc::Async.new(|@argv);230 my @stdout-chunks;231 my @stderr-chunks;232 $proc.stdout.tap(-> $chunk { @stdout-chunks.push($chunk) });233 $proc.stderr.tap(-> $chunk { @stderr-chunks.push($chunk) });234235 my $timeout-secs = $timeout.defined236 ?? $timeout.Numeric237 !! (%*ENV<BEHAVE_DISCOVERY_TIMEOUT> // 300).Numeric;238239 my $result;240 try {241 my $promise = $proc.start(:ENV(%env));242 await Promise.anyof($promise, Promise.in($timeout-secs));243244 unless $promise.status ~~ Kept {245 $proc.kill(SIGKILL);246 my $reaped = try await $promise;247 my $msg = "discovery subprocess timed out after {$timeout-secs}s "248 ~ "(a spec may hang while loading, e.g. a stale precompilation)";249 for @spec-files -> $f {250 @load-errors.push: %( file => $f.Str, :message($msg) );251 }252 return (@suites.List, @load-errors.List).List;253 }254255 $result = $promise.result;256 CATCH {257 default {258 for @spec-files -> $f {259 @load-errors.push: %( file => $f.Str, message => "discovery subprocess failed: {.message}" );260 }261 return (@suites.List, @load-errors.List).List;262 }263 }264 }265266 my $stdout = @stdout-chunks.join;267 my $stderr = @stderr-chunks.join;268269 if $result.signal || $result.exitcode > 1 {270 my $msg = $result.signal271 ?? "discovery subprocess died with signal {$result.signal}"272 ~ " (a corrupt precompilation cache can segfault the loader;"273 ~ " clearing the project's .precomp directories usually clears it)"274 !! "discovery subprocess exited with code {$result.exitcode}";275 $msg ~= ": $stderr" if $stderr.chars;276 for @spec-files -> $f {277 @load-errors.push: %( file => $f.Str, :message($msg) );278 }279 return (@suites.List, @load-errors.List).List;280 }281282 my %doc;283 try {284 %doc = BDD::Behave::Coverage::parse-baseline-json($stdout);285 CATCH {286 default {287 for @spec-files -> $f {288 @load-errors.push: %( file => $f.Str, message => "discovery subprocess returned invalid JSON: {.message}" );289 }290 return (@suites.List, @load-errors.List).List;291 }292 }293 }294295 for ((%doc<suites> // ()).list) -> %node {296 my $suite = rebuild-suite(%node);297 @suites.push: $suite if $suite.defined;298 }299300 for ((%doc<load-errors> // ()).list) -> %err {301 @load-errors.push: %(302 file => (%err<file> // '').Str,303 message => (%err<message> // '').Str,304 );305 }306307 (@suites.List, @load-errors.List).List;308}309310sub rebuild-suite(%node) {311 my %meta = json-metadata-to-raku(%node<metadata> // %());312 my $suite = Suite.new(313 :description((%node<description> // '').Str),314 :file((%node<file> // '').IO),315 :line((%node<line> // 0).Int),316 :metadata(%meta),317 );318 for ((%node<children> // ()).list) -> %child {319 rebuild-child-into($suite, %child);320 }321 $suite;322}323324sub rebuild-child-into($parent, %node) {325 my $type = (%node<type> // '').Str;326 my %meta = json-metadata-to-raku(%node<metadata> // %());327 given $type {328 when 'group' {329 my $group = ExampleGroup.new(330 :description((%node<description> // '').Str),331 :file((%node<file> // '').IO),332 :line((%node<line> // 0).Int),333 :metadata(%meta),334 );335 $parent.add-child($group);336 for ((%node<children> // ()).list) -> %child {337 rebuild-child-into($group, %child);338 }339 }340 when 'example' {341 my $example = Example.new(342 :description((%node<description> // '').Str),343 :file((%node<file> // '').IO),344 :line((%node<line> // 0).Int),345 :metadata(%meta),346 :block(sub { }),347 );348 $example.pending = ?(%node<pending>);349 $parent.add-child($example);350 }351 }352}353354sub json-metadata-to-raku(%raw --> Hash) {355 my %m;356 for %raw.kv -> $k, $v {357 %m{$k} = $v;358 }359 %m;360}361362sub locations-for-buckets(@buckets --> List) is export {363 my @all;364 for @buckets -> $b {365 @all.append: $b.locations.list;366 }367 @all.List;368}369370sub collect-filtered-buckets(371 @suites,372 :@include-tags,373 :@exclude-tags,374 :@example-patterns,375 :@only-locations,376 --> List377) is export {378 my @all-buckets;379 for @suites -> $suite {380 @all-buckets.append: collect-buckets($suite).list;381 }382383 my @filtered = @all-buckets.map(-> $b {384 my @kept = $b.examples.grep({385 example-passes-filters($_, :@include-tags, :@exclude-tags, :@example-patterns, :@only-locations);386 });387 if @kept.elems == $b.examples.elems {388 $b;389 } elsif @kept.elems == 0 {390 Nil;391 } else {392 my $new = BDD::Behave::Parallel::Distribution::Bucket.new(393 :id($b.id),394 :file($b.file),395 );396 $new.add($_) for @kept;397 $new.serial = $b.serial;398 $new;399 }400 }).grep(*.defined);401402 @filtered.List;403}404405sub build-worker-manifests(406 @suites,407 Int $worker-count,408 :@include-tags,409 :@exclude-tags,410 :@example-patterns,411 :@only-locations,412 Str :$seed-mode = 'xor',413 Int :$seed,414 --> Hash415) is export {416 my @filtered = collect-filtered-buckets(417 @suites,418 :@include-tags,419 :@exclude-tags,420 :@example-patterns,421 :@only-locations,422 );423424 my ($parallel-buckets, $serial-buckets) = split-parallel-and-serial(@filtered);425426 my @parallel-assignments = $seed-mode eq 'stable'427 ?? distribute-stable($parallel-buckets, $worker-count, $seed // 0)428 !! distribute-lpt($parallel-buckets, $worker-count);429430 my @parallel-manifests;431 for ^$worker-count -> $i {432 my @locs;433 for @parallel-assignments[$i].list -> $b {434 @locs.append: $b.locations.list;435 }436 @parallel-manifests.push: @locs;437 }438439 my @serial-locations;440 for $serial-buckets.list -> $b {441 @serial-locations.append: $b.locations.list;442 }443444 %(445 parallel-manifests => @parallel-manifests.List,446 serial-locations => @serial-locations.List,447 parallel-count => $parallel-buckets.list.map(*.cost).sum,448 serial-count => $serial-buckets.list.map(*.cost).sum,449 );450}451452sub example-passes-filters(Example $example, :@include-tags, :@exclude-tags, :@example-patterns, :@only-locations --> Bool) {453 my @tags = $example.effective-tags;454 if @exclude-tags.elems && @tags.first({ $_ ∈ @exclude-tags }).defined {455 return False;456 }457 if @include-tags.elems && !@tags.first({ $_ ∈ @include-tags }).defined {458 return False;459 }460 if @example-patterns.elems {461 my $desc = nested-description($example);462 my $matched = False;463 for @example-patterns -> $pat {464 if pattern-matches($desc, $pat) {465 $matched = True;466 last;467 }468 }469 return False unless $matched;470 }471 if @only-locations.elems {472 my $matched = False;473 my $ex-loc = "{$example.file.absolute}:{$example.line}";474 for @only-locations -> $loc {475 if location-matches($ex-loc, $loc) {476 $matched = True;477 last;478 }479 for $example.ancestry -> $node {480 next unless $node ~~ ExampleGroup;481 if location-matches("{$node.file.absolute}:{$node.line}", $loc) {482 $matched = True;483 last;484 }485 }486 last if $matched;487 }488 return False unless $matched;489 }490 True;491}492493sub nested-description(Example $example --> Str) {494 my @parts = $example.ancestry.grep(ExampleGroup).map(*.description);495 @parts.push($example.description);496 @parts.join(' ');497}498499sub pattern-matches(Str $description, Str $pattern --> Bool) {500 if $pattern.chars > 2 && $pattern.starts-with('/') && $pattern.ends-with('/') {501 my $body = $pattern.substr(1, $pattern.chars - 2);502 my $rx = / <{ $body }> /;503 return so $description.match($rx);504 }505 $description.contains($pattern);506}507508sub location-matches(Str $ex-loc, Str $pattern --> Bool) {509 return False unless $pattern.contains(':');510 my $idx = $pattern.rindex(':');511 my $pat-path = $pattern.substr(0, $idx);512 my $pat-line = $pattern.substr($idx + 1);513 my $ex-idx = $ex-loc.rindex(':');514 return False unless $ex-idx.defined;515 my $ex-path = $ex-loc.substr(0, $ex-idx);516 my $ex-line = $ex-loc.substr($ex-idx + 1);517 return False unless $ex-line eq $pat-line;518 return True if $ex-path eq $pat-path;519 return True if $ex-path.IO.absolute eq $pat-path.IO.absolute;520 return True if $ex-path.ends-with('/' ~ $pat-path);521 False;522}523524class ShardRetryRecord is export {525 has Int $.worker is required;526 has Int $.attempts is required;527 has Int $.final-exit is required;528 has Str $.outcome is required;529 has @.crash-codes;530}531532class ParallelRunResult is export {533 has Int $.total is rw = 0;534 has Int $.passed is rw = 0;535 has Int $.failed is rw = 0;536 has Int $.pending is rw = 0;537 has Int $.skipped is rw = 0;538 has @.failures;539 has @.load-errors;540 has @.retry-records;541 has @.shard-retry-records;542 has @.executed-locations;543 has @.failed-locations;544 has @.timed-examples;545 has @.benchmark-summaries;546 has Int $.exit-code is rw = 0;547548 method success(--> Bool) { $!failed == 0 && @!load-errors.elems == 0 && $!exit-code == 0 }549}550551sub run-parallel-isolated-impl(552 ParallelRunOptions $opts,553 ParallelRunResult $result,554 @suites,555 --> ParallelRunResult556) {557 my @filtered = collect-filtered-buckets(558 @suites,559 :include-tags($opts.include-tags),560 :exclude-tags($opts.exclude-tags),561 :example-patterns($opts.example-patterns),562 :only-locations($opts.only-locations),563 );564565 my @file-buckets = coalesce-by-file(@filtered);566567 if $opts.progress-total {568 my $total = @file-buckets.map(*.cost).sum;569 $opts.formatter.set-total($total);570 }571572 return $result unless @file-buckets.elems;573574 my $manifest-dir = $*TMPDIR.add("behave-parallel-isolated-{$*PID}-{(now * 1e6).Int}");575 $manifest-dir.mkdir;576577 LEAVE {578 if $manifest-dir.defined && $manifest-dir.e && $manifest-dir.d {579 for $manifest-dir.dir -> $f { $f.unlink if $f.f }580 $manifest-dir.rmdir;581 }582 }583584 my Lock $event-lock .= new;585586 # Build one job per spec file (manifest + argv + env), then feed them through587 # a closed Channel to a fixed pool of exactly worker-count threads. Each588 # thread owns a stable slot (0 .. N-1) that it exports as BEHAVE_WORKER_INDEX589 # and pulls one job at a time, so concurrently-running files always hold590 # distinct indices — user code can key a per-worker resource (e.g. a database591 # `myapp_test_{Worker.id}`) off the index with only N to provision. Using a592 # fixed N-thread pool (not one `start` per file) is essential: one start per593 # file would pile up worker-count-blocked threads and starve the thread pool594 # that Proc::Async needs to drain worker stdout, deadlocking the run.595 my $queue = Channel.new;596 my $file-index = 0;597598 for @file-buckets -> $bucket {599 my $idx = $file-index++;600 my $file = $bucket.file;601 my @locations = $bucket.locations.list;602 my $manifest-path = $manifest-dir.add("file-$idx.manifest");603 write-manifest($manifest-path, @locations);604605 my @argv = $opts.worker-argv.Slip;606 @argv.push: '--worker-manifest', $manifest-path.absolute;607 @argv.push: $file;608609 my %env = |$opts.base-env;610 %env<BEHAVE_WORKER_COUNT> = $opts.worker-count.Str;611612 if $opts.coverage-log-dir.defined {613 %env<MVM_COVERAGE_LOG>614 = $opts.coverage-log-dir.add("isolated-$idx.raw").absolute;615 %env<MVM_COVERAGE_CONTROL> = $opts.coverage-counts ?? '2' !! '0';616 }617618 $queue.send({ :$idx, :$file, :@argv, :%env });619 }620 $queue.close;621622 my @promises;623 for ^$opts.worker-count -> $slot {624 @promises.push: start {625 loop {626 my %job = $queue.receive;627 CATCH { when X::Channel::ReceiveOnClosed { last } }628629 my @argv = %job<argv>.list;630 my %env = %job<env>.hash;631 my $idx = %job<idx>;632 my $file = %job<file>;633634 %env<BEHAVE_WORKER_INDEX> = $slot.Str;635636 my $proc = Proc::Async.new(|@argv);637 my $parser = JsonLineParser.new;638639 $proc.stdout.tap(-> $chunk {640 for $parser.feed($chunk) -> $event {641 $event-lock.protect: {642 handle-event($opts.formatter, $result, $idx, $event, @suites);643 }644 }645 });646 $proc.stderr.tap(-> $chunk {647 $event-lock.protect: { $*ERR.print($chunk) }648 });649650 my $proc-result = await $proc.start(:ENV(%env));651652 for $parser.flush -> $event {653 $event-lock.protect: {654 handle-event($opts.formatter, $result, $idx, $event, @suites);655 }656 }657658 my $effective = $proc-result.signal659 ?? 128 + $proc-result.signal660 !! $proc-result.exitcode;661 if $effective > 1 {662 $event-lock.protect: {663 $result.exit-code = 1;664 note red($proc-result.signal665 ?? "Isolated worker for $file died with signal {$proc-result.signal}"666 !! "Isolated worker for $file exited with code {$proc-result.exitcode}");667 }668 }669 }670 };671 }672673 await Promise.allof(@promises);674 $result;675}676677sub run-parallel-queue-impl(678 ParallelRunOptions $opts,679 ParallelRunResult $result,680 @suites,681 --> ParallelRunResult682) {683 my @filtered = collect-filtered-buckets(684 @suites,685 :include-tags($opts.include-tags),686 :exclude-tags($opts.exclude-tags),687 :example-patterns($opts.example-patterns),688 :only-locations($opts.only-locations),689 );690691 my ($parallel-buckets, $serial-buckets) = split-parallel-and-serial(@filtered);692 my @parallel = $parallel-buckets.list;693 my @serial = $serial-buckets.list;694695 if $opts.progress-total {696 my $total = @parallel.map(*.cost).sum + @serial.map(*.cost).sum;697 $opts.formatter.set-total($total);698 }699700 if @parallel.elems {701 my $worker-count = $opts.worker-count min @parallel.elems;702 my $scheduler = BDD::Behave::Parallel::Queue::QueueScheduler.new;703 $scheduler.enqueue-sorted(@parallel);704705 my Lock $dispatch-lock .= new;706 my Lock $event-lock .= new;707 my $pool;708709 my &dispatch-or-shutdown = sub (Int $wi --> Nil) {710 my $next;711 $dispatch-lock.protect: {712 $next = $scheduler.next-bucket;713 }714 if $next.defined {715 $pool.send-bucket($wi, $next);716 } else {717 $pool.send-shutdown($wi);718 }719 }720721 $pool = BDD::Behave::Parallel::Queue::QueueWorkerPool.new(722 :$worker-count,723 :worker-argv($opts.worker-argv),724 :spec-files($opts.spec-files.map(*.Str).List),725 :base-env(%(|$opts.base-env)),726 :coverage-log-dir($opts.coverage-log-dir),727 :coverage-counts($opts.coverage-counts),728 # Tap callbacks for every worker fire on the thread pool, so729 # several can be processing events concurrently. handle-event730 # mutates $result.total / $result.passed / ... and writes to the731 # parent formatter; both are unsynchronized state, so we serialize732 # all per-event work behind a single lock.733 :on-event(sub ($wi, $event) {734 $event-lock.protect: {735 handle-event($opts.formatter, $result, $wi, $event, @suites);736 }737 }),738 :on-ready(&dispatch-or-shutdown),739 :on-done(sub ($wi, $id) {740 $dispatch-lock.protect: { $scheduler.mark-complete }741 dispatch-or-shutdown($wi);742 }),743 );744745 $pool.launch;746 $pool.wait-all;747 for $pool.workers -> $w {748 if $w.exit-code > 1 {749 $result.exit-code = 1;750 note red("Queue worker {$w.index} exited with code {$w.exit-code}");751 }752 }753 }754755 if @serial.elems {756 my $serial-dir = $*TMPDIR.add("behave-parallel-queue-serial-{$*PID}-{(now * 1e6).Int}");757 $serial-dir.mkdir;758 LEAVE {759 if $serial-dir.defined && $serial-dir.e && $serial-dir.d {760 for $serial-dir.dir -> $f { $f.unlink if $f.f }761 $serial-dir.rmdir;762 }763 }764765 my @serial-locations;766 for @serial -> $b { @serial-locations.append: $b.locations.list }767768 my $serial-manifest-path = $serial-dir.add('serial.manifest');769 write-manifest($serial-manifest-path, @serial-locations);770 my @files = files-from-manifest(@serial-locations);771 my @argv = $opts.worker-argv.Slip;772 @argv.push: '--worker-manifest', $serial-manifest-path.absolute;773 @argv.append: @files;774775 my %env = |$opts.base-env;776 %env<BEHAVE_WORKER_INDEX> = '0';777 %env<BEHAVE_WORKER_COUNT> = '1';778779 if $opts.coverage-log-dir.defined {780 %env<MVM_COVERAGE_LOG>781 = $opts.coverage-log-dir.add('serial.raw').absolute;782 %env<MVM_COVERAGE_CONTROL> = $opts.coverage-counts ?? '2' !! '0';783 }784785 my $proc = Proc::Async.new(|@argv);786 my $parser = JsonLineParser.new;787788 $proc.stdout.tap(-> $chunk {789 for $parser.feed($chunk) -> $event {790 handle-event($opts.formatter, $result, -1, $event, @suites);791 }792 });793 $proc.stderr.tap(-> $chunk {794 $*ERR.print($chunk);795 });796797 my $start = $proc.start(:ENV(%env));798 my $proc-result = await $start;799 for $parser.flush -> $event {800 handle-event($opts.formatter, $result, -1, $event, @suites);801 }802 my $effective = $proc-result.signal803 ?? 128 + $proc-result.signal804 !! $proc-result.exitcode;805 if $effective > 1 {806 $result.exit-code = 1;807 note red($proc-result.signal808 ?? "Serial worker died with signal {$proc-result.signal}"809 !! "Serial worker exited with code {$proc-result.exitcode}");810 }811 }812813 $result;814}815816sub run-parallel(817 ParallelRunOptions $opts,818 --> ParallelRunResult819) is export {820 my $result = ParallelRunResult.new;821822 precompile-specs-serially(823 $opts.spec-files,824 :discovery-argv($opts.discovery-argv),825 :base-env(%(|$opts.base-env)),826 ) if should-precompile(827 :file-count($opts.spec-files.elems),828 :requested($opts.precompile-pass),829 :discovery-in-process($opts.discovery-in-process),830 );831832 my $disco = $opts.discovery-in-process833 ?? discover-suites($opts.spec-files)834 !! discover-suites-subprocess(835 $opts.spec-files,836 :discovery-argv($opts.discovery-argv),837 :base-env(%(|$opts.base-env)),838 );839 my @suites = $disco[0].list;840 my @load-errors = $disco[1].list;841 $result.load-errors.append: @load-errors;842843 my $mode = BDD::Behave::Configuration::resolve-parallel-mode($opts.parallel-mode);844845 if $mode eq 'queue' {846 return run-parallel-queue-impl($opts, $result, @suites);847 }848849 if $mode eq 'isolated' {850 return run-parallel-isolated-impl($opts, $result, @suites);851 }852853 my %plan = build-worker-manifests(854 @suites,855 $opts.worker-count,856 :include-tags($opts.include-tags),857 :exclude-tags($opts.exclude-tags),858 :example-patterns($opts.example-patterns),859 :only-locations($opts.only-locations),860 :seed-mode($opts.seed-mode),861 :seed($opts.seed),862 );863864 my @parallel-manifests = %plan<parallel-manifests>.list;865 my @serial-locations = %plan<serial-locations>.list;866867 if $opts.progress-total {868 my $total = (%plan<parallel-count> // 0).Int + (%plan<serial-count> // 0).Int;869 $opts.formatter.set-total($total);870 }871872 my $manifest-dir = $*TMPDIR.add("behave-parallel-{$*PID}-{(now * 1e6).Int}");873 $manifest-dir.mkdir;874875 LEAVE {876 if $manifest-dir.defined && $manifest-dir.e && $manifest-dir.d {877 for $manifest-dir.dir -> $f { $f.unlink if $f.f }878 $manifest-dir.rmdir;879 }880 }881882 if @parallel-manifests.grep({ .elems }).elems {883 my Lock $event-lock .= new;884 my $pool = WorkerPool.new(885 :worker-count($opts.worker-count),886 :worker-argv($opts.worker-argv),887 :base-env(%(|$opts.base-env)),888 :manifest-dir($manifest-dir),889 :coverage-log-dir($opts.coverage-log-dir),890 :coverage-counts($opts.coverage-counts),891 :retry-count($opts.parallel-retry),892 # Tap callbacks fire on the thread pool — N workers can dispatch893 # concurrently into handle-event, which mutates $result.* and894 # writes to the parent formatter. Serialize both.895 :on-event(sub ($wi, $event) {896 $event-lock.protect: {897 handle-event($opts.formatter, $result, $wi, $event, @suites);898 }899 }),900 :on-shard-retry(sub ($wi, $attempt, $exit-code) {901 $event-lock.protect: {902 note yellow("Worker $wi crashed with exit $exit-code; retrying (attempt $attempt of {$opts.parallel-retry + 1})");903 }904 }),905 );906 $pool.launch(@parallel-manifests);907 $pool.wait-all;908 for $pool.workers -> $w {909 if $w.attempt > 1 {910 my $outcome = $w.exit-code > 1 ?? 'crashed' !! 'recovered';911 $result.shard-retry-records.push: ShardRetryRecord.new(912 :worker($w.index),913 :attempts($w.attempt),914 :final-exit($w.exit-code),915 :$outcome,916 :crash-codes($w.crash-codes.list),917 );918 }919 if $w.exit-code > 1 {920 $result.exit-code = 1;921 note red("Worker {$w.index} exited with code {$w.exit-code} (after {$w.attempt} attempt{$w.attempt == 1 ?? '' !! 's'})");922 }923 }924 }925926 if @serial-locations.elems {927 my $serial-manifest-path = $manifest-dir.add('serial.manifest');928 write-manifest($serial-manifest-path, @serial-locations);929 my @files = files-from-manifest(@serial-locations);930 my @argv = $opts.worker-argv.Slip;931 @argv.push: '--worker-manifest', $serial-manifest-path.absolute;932 @argv.append: @files;933934 my %env = |$opts.base-env;935 %env<BEHAVE_WORKER_INDEX> = '0';936 %env<BEHAVE_WORKER_COUNT> = '1';937938 if $opts.coverage-log-dir.defined {939 %env<MVM_COVERAGE_LOG>940 = $opts.coverage-log-dir.add('serial.raw').absolute;941 %env<MVM_COVERAGE_CONTROL> = $opts.coverage-counts ?? '2' !! '0';942 }943944 my $proc = Proc::Async.new(|@argv);945 my $parser = JsonLineParser.new;946947 $proc.stdout.tap(-> $chunk {948 for $parser.feed($chunk) -> $event {949 handle-event($opts.formatter, $result, -1, $event, @suites);950 }951 });952 $proc.stderr.tap(-> $chunk {953 $*ERR.print($chunk);954 });955956 my $start = $proc.start(:ENV(%env));957 my $proc-result = await $start;958 for $parser.flush -> $event {959 handle-event($opts.formatter, $result, -1, $event, @suites);960 }961 my $effective = $proc-result.signal962 ?? 128 + $proc-result.signal963 !! $proc-result.exitcode;964 if $effective > 1 {965 $result.exit-code = 1;966 note red($proc-result.signal967 ?? "Serial worker died with signal {$proc-result.signal}"968 !! "Serial worker exited with code {$proc-result.exitcode}");969 }970 $serial-manifest-path.unlink if $serial-manifest-path.e;971 }972973 $result;974}975976sub handle-event($formatter, ParallelRunResult $result, Int $worker, %event, @suites) {977 my $type = %event<type> // 'unknown';978 given $type {979 when 'suite-start' {980 my $suite = lookup-suite(@suites, %event<id>);981 $formatter.suite-start($suite, :multi-file(@suites.elems > 1)) if $suite.defined;982 }983 when 'suite-end' {984 my $suite = lookup-suite(@suites, %event<id>);985 $formatter.suite-end($suite) if $suite.defined;986 }987 when 'group-start' {988 my $group = lookup-group(@suites, %event<id>);989 $formatter.group-start($group) if $group.defined;990 }991 when 'group-end' {992 my $group = lookup-group(@suites, %event<id>);993 $formatter.group-end($group) if $group.defined;994 }995 when 'group-around-skipped' {996 my $group = lookup-group(@suites, %event<id>);997 $formatter.group-around-skipped($group) if $group.defined;998 }999 when 'example-start' {1000 my $example = resolve-example(@suites, %event);1001 if $example.defined {1002 $example.started-at = now;1003 $formatter.example-start($example, :auto(?%event<auto>));1004 }1005 }1006 when 'example-auto-description' {1007 my $example = resolve-example(@suites, %event);1008 $formatter.example-auto-description($example, :description(%event<description> // ''))1009 if $example.defined;1010 }1011 when 'example-pass' {1012 my $example = resolve-example(@suites, %event);1013 if $example.defined {1014 $example.duration = (%event<duration> // 0).Real;1015 $formatter.example-pass($example);1016 $result.executed-locations.push("{$example.file.absolute}:{$example.line}");1017 }1018 $result.total++;1019 $result.passed++;1020 }1021 when 'example-fail' {1022 my $example = resolve-example(@suites, %event);1023 my %failure-info = (1024 description => (%event<failure-description> // ($example.defined ?? $example.description !! '')),1025 file => (%event<failure-file> // ($example.defined ?? $example.file.absolute !! '')),1026 line => (%event<failure-line> // ($example.defined ?? $example.line !! 0)),1027 );1028 with %event<exception-message> {1029 %failure-info<exception-message> = $_;1030 }1031 with %event<exception-backtrace> {1032 %failure-info<exception-backtrace> = $_;1033 }10341035 my @expectations = (%event<failures> // ()).list.map(-> %f {1036 my %rec = (1037 file => (%f<file> // '').Str,1038 line => (%f<line> // 0).Int,1039 negated => (%f<negated> // False),1040 given => (%f<given> // '').Str,1041 expected => (%f<expected> // '').Str,1042 );1043 %rec<message> = %f<message>.Str if (%f<message> // '').Str.chars;1044 %rec;1045 });1046 %failure-info<expectations> = @expectations if @expectations.elems;10471048 $result.total++;1049 $result.failed++;1050 $result.failures.push: %failure-info;10511052 # Record this example's failures on the global list before emitting the1053 # event. A streaming formatter derives an example's failures from the list1054 # since the previous event, so recording after emitting would attribute1055 # them to the next example instead of this one.1056 my $desc = %failure-info<description>;1057 my @event-failures = (%event<failures> // ()).list;1058 for @event-failures -> %f {1059 my $given = (%f<given> // '').Str;1060 my $expected = (%f<expected> // '').Str;1061 my $negated = ?(%f<negated> // False);1062 my $msg = (%f<message> // '').Str;1063 $msg = compose-failure-message($given, $expected, $negated) unless $msg.chars;10641065 Failures.list.push: Failure.new(1066 :file((%f<file> // '').Str),1067 :line((%f<line> // 0).Int),1068 :given($given.chars ?? $given !! Nil),1069 :expected($expected.chars ?? $expected !! Nil),1070 :$negated,1071 :message($msg.chars ?? $msg !! Str),1072 :aggregation-label(1073 ((%f<aggregation-label> // '').Str.chars1074 ?? %f<aggregation-label>.Str !! Str)),1075 :description($desc),1076 );1077 }1078 unless @event-failures.elems {1079 with %event<exception-message> -> $msg {1080 my $prefixed = $desc.chars1081 ?? "exception in $desc: " ~ $msg.Str1082 !! $msg.Str;10831084 Failures.list.push: Failure.new(1085 :file(%failure-info<file>.Str),1086 :line(%failure-info<line>.Int),1087 :message($prefixed),1088 :description($desc),1089 :from-runner-exception,1090 );1091 }1092 }10931094 if $example.defined {1095 $example.duration = (%event<duration> // 0).Real;1096 $formatter.example-fail($example, :%failure-info);1097 my $loc = "{$example.file.absolute}:{$example.line}";1098 $result.executed-locations.push($loc);1099 $result.failed-locations.push($loc);1100 } else {1101 my $f = %failure-info<file>.Str;1102 my $l = %failure-info<line>.Int;1103 $result.failed-locations.push("$f:$l") if $f.chars && $l;1104 }1105 }1106 when 'example-retry' {1107 my $example = resolve-example(@suites, %event);1108 $formatter.example-retry(1109 $example,1110 :attempt((%event<attempt> // 1).Int),1111 :max-attempts((%event<max-attempts> // 1).Int),1112 ) if $example.defined;1113 }1114 when 'example-pending' {1115 my $example = resolve-example(@suites, %event);1116 $formatter.example-pending($example) if $example.defined;1117 $result.total++;1118 $result.pending++;1119 }1120 when 'example-skipped' {1121 my $example = resolve-example(@suites, %event);1122 $formatter.example-skipped($example) if $example.defined;1123 $result.total++;1124 $result.skipped++;1125 }1126 when 'example-around-skipped' {1127 my $example = resolve-example(@suites, %event);1128 $formatter.example-around-skipped($example) if $example.defined;1129 $result.total++;1130 $result.skipped++;1131 }1132 when 'example-slow' {1133 my $example = resolve-example(@suites, %event);1134 $formatter.example-slow($example, :threshold((%event<threshold> // 0).Real))1135 if $example.defined;1136 }1137 when 'profile-record' {1138 my $example = resolve-example(@suites, %event);1139 $result.timed-examples.push: %(1140 example => $example,1141 description => (%event<description> // '').Str,1142 duration => (%event<duration> // 0).Real,1143 );1144 }1145 when 'benchmark-record' {1146 my $example = resolve-example(@suites, %event);1147 my @timings = (%event<timings> // ()).list.map(*.Real).List;1148 $result.benchmark-summaries.push: %(1149 example => $example,1150 description => (%event<description> // '').Str,1151 key => (%event<key> // '').Str,1152 label => (%event<label>.defined && (%event<label>).Str.chars1153 ?? %event<label>.Str !! Str),1154 position => (%event<position> // 0).Int,1155 runs => (%event<runs> // 0).Int,1156 iterations => (%event<iterations> // 0).Int,1157 timings => @timings,1158 min => (%event<min> // 0).Real,1159 max => (%event<max> // 0).Real,1160 mean => (%event<mean> // 0).Real,1161 median => (%event<median> // 0).Real,1162 total => (%event<total> // 0).Real,1163 );1164 }1165 when 'retry-record' {1166 my $rec = BDD::Behave::Runner::RetryRecord.new(1167 :description((%event<description> // '').Str),1168 :location((%event<location> // '').Str),1169 :attempts((%event<attempts> // 1).Int),1170 :max-attempts((%event<max-attempts> // 1).Int),1171 :outcome((%event<outcome> // 'pass').Str),1172 );1173 $result.retry-records.push($rec);1174 }1175 when 'load-error' {1176 $result.load-errors.push: %( file => %event<file>, message => %event<message> );1177 }1178 when 'parse-error' {1179 note red("Worker $worker emitted unparseable event: " ~ (%event<raw> // ''));1180 }1181 when 'run-summary' { }1182 default { }1183 }1184}11851186sub lookup-suite(@suites, Str $id) {1187 return Nil unless $id.defined;1188 for @suites -> $suite {1189 return $suite if "{$suite.file.absolute}:{$suite.line}" eq $id;1190 }1191 Nil;1192}11931194sub lookup-group(@suites, Str $id) {1195 return Nil unless $id.defined;1196 for @suites -> $suite {1197 my $hit = walk-find-group($suite, $id);1198 return $hit if $hit.defined;1199 }1200 Nil;1201}12021203sub walk-find-group($container, Str $id) {1204 for $container.children -> $child {1205 given $child {1206 when ExampleGroup {1207 return $child if "{$child.file.absolute}:{$child.line}" eq $id;1208 my $deeper = walk-find-group($child, $id);1209 return $deeper if $deeper.defined;1210 }1211 }1212 }1213 Nil;1214}12151216sub lookup-example(@suites, Str $id) {1217 return Nil unless $id.defined;1218 for @suites -> $suite {1219 my $hit = walk-find-example($suite, $id);1220 return $hit if $hit.defined;1221 }1222 Nil;1223}12241225sub resolve-example(@suites, %event) {1226 my $hit = lookup-example(@suites, %event<id>);1227 return $hit if $hit.defined;1228 my $desc = (%event<description> // %event<failure-description> // '').Str;1229 my $file = (%event<file> // %event<failure-file> // '').Str;1230 my $line = (%event<line> // %event<failure-line> // 0).Int;1231 return Nil unless $desc.chars || $file.chars;1232 Example.new(1233 :description($desc),1234 :file($file.IO),1235 :line($line),1236 :block({ }),1237 );1238}12391240sub walk-find-example($container, Str $id) {1241 for $container.children -> $child {1242 given $child {1243 when Example {1244 return $child if "{$child.file.absolute}:{$child.line}" eq $id;1245 }1246 when ExampleGroup {1247 my $deeper = walk-find-example($child, $id);1248 return $deeper if $deeper.defined;1249 }1250 }1251 }1252 Nil;1253}