68 / 73 lines covered (93.2%)
1unit module BDD::Behave::Parallel::WorkerPool;23use BDD::Behave::Parallel::EventStream;4use BDD::Behave::Parallel::Manifest;56class WorkerHandle is export {7 has Int $.index is required;8 has Proc::Async $.proc is rw;9 has IO::Path $.manifest-path is required;10 has Promise $.exit-promise is required;11 has JsonLineParser $.parser is rw = JsonLineParser.new;12 has @.events is rw;13 has Bool $.finished is rw = False;14 has Int $.exit-code is rw = 0;15 # Chunks are kept as they arrive and joined only when someone asks, so a16 # worker printing a lot of stderr does not re-copy what it already printed.17 has @.stderr-chunks is rw;1819 method stderr-output(--> Str) { @!stderr-chunks.join }20 has Int $.attempt is rw = 1;21 has @.crash-codes is rw;22}2324class WorkerPool is export {25 has Int $.worker-count is required;26 has @.worker-argv is required;27 has %.base-env;28 has IO::Path $.manifest-dir is required;29 has IO::Path $.coverage-log-dir;30 has Bool $.coverage-counts = False;31 has Int $.retry-count = 0;32 has &.on-event = sub ($wi, $event) { };33 has &.on-shard-retry = sub ($wi, $attempt, $exit-code) { };3435 has WorkerHandle @!workers;36 has @!worker-manifests;3738 method launch(@manifests --> Nil) {39 die "manifest count {@manifests.elems} must equal worker-count $!worker-count"40 unless @manifests.elems == $!worker-count;4142 @!worker-manifests = @manifests;4344 for ^$!worker-count -> $i {45 my @locations = @manifests[$i].List;46 my $manifest-path = $!manifest-dir.add("worker-$i.manifest");47 write-manifest($manifest-path, @locations);4849 my $handle = WorkerHandle.new(50 :index($i),51 :proc(Proc::Async),52 :$manifest-path,53 :exit-promise(Promise.new),54 );5556 @!workers.push($handle);57 self!spawn-attempt($handle);58 }59 }6061 method !spawn-attempt(WorkerHandle $handle --> Nil) {62 my @locations = read-manifest($handle.manifest-path);63 my @files = files-from-manifest(@locations);6465 my @argv = @!worker-argv.Slip;66 @argv.push: '--worker-manifest', $handle.manifest-path.absolute;67 @argv.append: @files;6869 my %env = |%!base-env;70 %env<BEHAVE_WORKER_INDEX> = $handle.index.Str;71 %env<BEHAVE_WORKER_COUNT> = $!worker-count.Str;7273 if $!coverage-log-dir.defined {74 %env<MVM_COVERAGE_LOG>75 = $!coverage-log-dir.add("worker-{$handle.index}.raw").absolute;76 %env<MVM_COVERAGE_CONTROL> = $!coverage-counts ?? '2' !! '0';77 }7879 my $proc = Proc::Async.new(|@argv);80 $handle.proc = $proc;81 $handle.parser = JsonLineParser.new;82 $handle.events = [];83 $handle.stderr-chunks = [];8485 my $buffering = $!retry-count > 0;8687 $proc.stdout.tap(-> $chunk {88 try {89 my @events = $handle.parser.feed($chunk);90 for @events -> $event {91 $handle.events.push($event);92 &!on-event($handle.index, $event) unless $buffering;93 }94 CATCH { default { note "Worker {$handle.index} stdout tap error: {.message}" } }95 }96 });9798 $proc.stderr.tap(-> $chunk {99 $handle.stderr-chunks.push($chunk);100 $*ERR.print($chunk);101 });102103 my $start-promise = $proc.start(:ENV(%env));104 $start-promise.then({105 my $retried = False;106 try {107 my $proc-result = $start-promise.result;108 # A worker killed by a signal (e.g. SIGSEGV) reports exitcode 0 and109 # the signal separately; fold it to the 128+N shell convention so110 # every crash check downstream sees it as a crash, never as success.111 my $exit-code = $proc-result.signal112 ?? 128 + $proc-result.signal113 !! $proc-result.exitcode;114 note "Worker {$handle.index} died with signal {$proc-result.signal}"115 if $proc-result.signal;116 my @tail = $handle.parser.flush;117 for @tail -> $event {118 $handle.events.push($event);119 &!on-event($handle.index, $event) unless $buffering;120 }121122 if $exit-code > 1 && $handle.attempt <= $!retry-count {123 $handle.crash-codes.push: $exit-code;124 $handle.attempt++;125 &!on-shard-retry($handle.index, $handle.attempt, $exit-code);126 $retried = True;127 self!spawn-attempt($handle);128 } else {129 $handle.exit-code = $exit-code;130 if $buffering {131 for $handle.events -> $event {132 &!on-event($handle.index, $event);133 }134 }135 }136 CATCH { default { note "Worker {$handle.index} exit handler error: {.message}" } }137 }138 unless $retried {139 $handle.finished = True;140 $handle.exit-promise.keep($handle.exit-code);141 }142 });143 }144145 method wait-all(--> Nil) {146 await Promise.allof(@!workers.map(*.exit-promise));147 }148149 method workers(--> List) {150 @!workers.List;151 }152153 method any-nonzero-exit(--> Bool) {154 @!workers.first({ .exit-code != 0 }).defined;155 }156157 method cleanup-manifests {158 for @!workers -> $w {159 $w.manifest-path.unlink if $w.manifest-path.e;160 }161 }162}