Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Issue 1975: Fix WIN32 Proc::Async process termination issue #1995

Closed
wants to merge 4 commits into from
Closed

Issue 1975: Fix WIN32 Proc::Async process termination issue #1995

Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
569cbeb
Change 'supply' names to 'supplier' for clarity
jstuder-gh Jun 30, 2018
d9f874c
Fix WIN32 Proc::Async process termination issue
jstuder-gh Jun 30, 2018
f1e9ede
Fix potential race condition in Proc::Async start
jstuder-gh Oct 25, 2018
be1019e
Remove anonymous class in Proc::Async
jstuder-gh Oct 25, 2018
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
115 changes: 71 additions & 44 deletions src/core/Proc/Async.pm6
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -103,11 +103,11 @@ my class Proc::Async {
has $.enc = 'utf8';
has $.translate-nl = True;
has Bool $.started = False;
has $!stdout_supply;
has $!stdout_supplier;
has CharsOrBytes $!stdout_type;
has $!stderr_supply;
has $!stderr_supplier;
has CharsOrBytes $!stderr_type;
has $!merge_supply;
has $!merge_supplier;
has CharsOrBytes $!merge_type;
has $!stdin-fd;
has $!stdout-fd;
Expand All @@ -133,22 +133,36 @@ my class Proc::Async {
-> { (channel == 1 ?? $!stdout_descriptor_used !! $!stderr_descriptor_used).keep(True) }
}

method !pipe(\what, \the-supply, \type, \value, \fd-vow, \permit-channel) {
my class ProcInternalSupplier is Supplier::Preserving {
has $.vow is rw;

method done(::?CLASS:D: --> Nil) {
$!vow.keep(self) if $!vow;
self.Supplier::Preserving::done();
}

method quit(::?CLASS:D: $err) {
$!vow.keep( (self, $err) ) if $!vow;
self.Supplier::Preserving::quit($err);
}
}

method !pipe(\what, \the-supplier, \type, \value, \fd-vow, \permit-channel) {
X::Proc::Async::TapBeforeSpawn.new(handle => what, proc => self).throw
if $!started;
X::Proc::Async::CharsOrBytes.new(handle => what, proc => self).throw
if the-supply and type != value;
if the-supplier and type != value;

type = value;
the-supply //= Supplier::Preserving.new;
type = value;
the-supplier //= ProcInternalSupplier.new;

if nqp::iscont(fd-vow) {
my $native-descriptor = Promise.new;
fd-vow = $native-descriptor.vow;
Pipe.new(the-supply.Supply.Tappable, $native-descriptor, |self!pipe-cbs(permit-channel))
Pipe.new(the-supplier.Supply.Tappable, $native-descriptor, |self!pipe-cbs(permit-channel))
}
else {
the-supply.Supply
the-supplier.Supply
}
}

Expand All @@ -167,59 +181,59 @@ my class Proc::Async {

proto method stdout(|) {*}
multi method stdout(Proc::Async:D: :$bin!) {
die X::Proc::Async::SupplyOrStd.new if $!merge_supply;
die X::Proc::Async::SupplyOrStd.new if $!merge_supplier;
die X::Proc::Async::BindOrUse.new(:handle<stdout>, :use('get the stdout Supply'))
if $!stdout-fd;
$bin
?? self!pipe('stdout', $!stdout_supply, $!stdout_type, Bytes, $!stdout_descriptor_vow, 1)
?? self!pipe('stdout', $!stdout_supplier, $!stdout_type, Bytes, $!stdout_descriptor_vow, 1)
!! self.stdout(|%_)
}
multi method stdout(Proc::Async:D: :$enc, :$translate-nl) {
die X::Proc::Async::SupplyOrStd.new if $!merge_supply;
die X::Proc::Async::SupplyOrStd.new if $!merge_supplier;
die X::Proc::Async::BindOrUse.new(:handle<stdout>, :use('get the stdout Supply'))
if $!stdout-fd;
self!wrap-decoder:
self!pipe('stdout', $!stdout_supply, $!stdout_type, Chars, Nil, 1),
self!pipe('stdout', $!stdout_supplier, $!stdout_type, Chars, Nil, 1),
$enc, $!stdout_descriptor_vow, 1, :$translate-nl
}

proto method stderr(|) {*}
multi method stderr(Proc::Async:D: :$bin!) {
die X::Proc::Async::SupplyOrStd.new if $!merge_supply;
die X::Proc::Async::SupplyOrStd.new if $!merge_supplier;
die X::Proc::Async::BindOrUse.new(:handle<stderr>, :use('get the stderr Supply'))
if $!stderr-fd;
$bin
?? self!pipe('stderr', $!stderr_supply, $!stderr_type, Bytes, $!stderr_descriptor_vow, 2)
?? self!pipe('stderr', $!stderr_supplier, $!stderr_type, Bytes, $!stderr_descriptor_vow, 2)
!! self.stderr(|%_)
}
multi method stderr(Proc::Async:D: :$enc, :$translate-nl) {
die X::Proc::Async::SupplyOrStd.new if $!merge_supply;
die X::Proc::Async::SupplyOrStd.new if $!merge_supplier;
die X::Proc::Async::BindOrUse.new(:handle<stderr>, :use('get the stderr Supply'))
if $!stderr-fd;
self!wrap-decoder:
self!pipe('stderr', $!stderr_supply, $!stderr_type, Chars, Nil, 2),
self!pipe('stderr', $!stderr_supplier, $!stderr_type, Chars, Nil, 2),
$enc, $!stderr_descriptor_vow, 2, :$translate-nl
}

proto method Supply(|) {*}
multi method Supply(Proc::Async:D: :$bin!) {
die X::Proc::Async::SupplyOrStd.new if $!stdout_supply || $!stderr_supply;
die X::Proc::Async::SupplyOrStd.new if $!stdout_supplier || $!stderr_supplier;
die X::Proc::Async::BindOrUse.new(:handle<stdout>, :use('get the output Supply'))
if $!stdout-fd;
die X::Proc::Async::BindOrUse.new(:handle<stderr>, :use('get the output Supply'))
if $!stderr-fd;
$bin
?? self!pipe('merge', $!merge_supply, $!merge_type, Bytes, Nil, 0)
?? self!pipe('merge', $!merge_supplier, $!merge_type, Bytes, Nil, 0)
!! self.Supply(|%_)
}
multi method Supply(Proc::Async:D: :$enc, :$translate-nl) {
die X::Proc::Async::SupplyOrStd.new if $!stdout_supply || $!stderr_supply;
die X::Proc::Async::SupplyOrStd.new if $!stdout_supplier || $!stderr_supplier;
die X::Proc::Async::BindOrUse.new(:handle<stdout>, :use('get the output Supply'))
if $!stdout-fd;
die X::Proc::Async::BindOrUse.new(:handle<stderr>, :use('get the output Supply'))
if $!stderr-fd;
self!wrap-decoder:
self!pipe('merge', $!merge_supply, $!merge_type, Chars, Nil, 0),
self!pipe('merge', $!merge_supplier, $!merge_type, Chars, Nil, 0),
$enc, Nil, 0, :$translate-nl
}

Expand All @@ -236,17 +250,17 @@ my class Proc::Async {

method bind-stdout(IO::Handle:D $handle --> Nil) {
die X::Proc::Async::BindOrUse.new(:handle<stdout>, :use('get the stdout Supply'))
if $!stdout_supply;
if $!stdout_supplier;
die X::Proc::Async::BindOrUse.new(:handle<stdout>, :use('get the output Supply'))
if $!merge_supply;
if $!merge_supplier;
$!stdout-fd := $handle.native-descriptor;
}

method bind-stderr(IO::Handle:D $handle --> Nil) {
die X::Proc::Async::BindOrUse.new(:handle<stderr>, :use('get the stderr Supply'))
if $!stderr_supply;
if $!stderr_supplier;
die X::Proc::Async::BindOrUse.new(:handle<stderr>, :use('get the output Supply'))
if $!merge_supply;
if $!merge_supplier;
$!stderr-fd := $handle.native-descriptor;
}

Expand All @@ -258,30 +272,34 @@ my class Proc::Async {
$!ready_promise
}

method !capture(\callbacks,\std,\the-supply) {
method !capture(\callbacks,\std,\the-supplier) {
my $promise = Promise.new;
my $vow = $promise.vow;
the-supplier.vow = $promise.vow;
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This seems race-y ( although I haven't bothered to investigate that hunch )

Copy link
Contributor Author

@jstuder-gh jstuder-gh Oct 25, 2018 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Looking at it again, you're right. I figure this particular method is probably alright (since it's a private method called with 3 different suppliers each time), the 'start' method could definitely cause a data race if called multiple times from different threads. I've updated the branch to add a commit wrapping it in a Lock.

I've cherry-picked that commit and will submit it on it's own, so that it can be considered regardless of what we decide to do with this PR.

EDIT:
Submitted as PR #2425

my $ss = Rakudo::Internals::SupplySequencer.new(
on-data-ready => -> \data { the-supply.emit(data) },
on-completed => -> { the-supply.done(); $vow.keep(the-supply) },
on-error => -> \err { the-supply.quit(err); $vow.keep((the-supply,err)) });
on-data-ready => -> \data { the-supplier.emit(data) },
on-completed => -> { the-supplier.done() },
on-error => -> \err { the-supplier.quit(err) });
nqp::bindkey(callbacks,
std ~ '_bytes' ,
-> Mu \seq, Mu \data, Mu \err { $ss.process(seq, data, err) });
$promise;
}

has $!start-lock = Lock.new;
method start(Proc::Async:D: :$scheduler = $*SCHEDULER, :$ENV, :$cwd = $*CWD) {
X::Proc::Async::AlreadyStarted.new(proc => self).throw if $!started;
$!started = True;
$!start-lock.protect: {
X::Proc::Async::AlreadyStarted.new(proc => self).throw if $!started;
$!started = True;

my @blockers;
if $!stdin-fd ~~ Promise {
@blockers.push($!stdin-fd.then({ $!stdin-fd := .result }));
my @blockers;
if $!stdin-fd ~~ Promise {
@blockers.push($!stdin-fd.then({ $!stdin-fd := .result }));
}
@blockers
?? start { await @blockers; await self!start-internal(:$scheduler, :$ENV, :$cwd) }
!! self!start-internal(:$scheduler, :$ENV, :$cwd)
}
@blockers
?? start { await @blockers; await self!start-internal(:$scheduler, :$ENV, :$cwd) }
!! self!start-internal(:$scheduler, :$ENV, :$cwd)
}

method !start-internal(:$scheduler, :$ENV, :$cwd) {
Expand Down Expand Up @@ -322,16 +340,16 @@ my class Proc::Async {
});

@!promises.push(Promise.anyof(
self!capture($callbacks,'stdout',$!stdout_supply),
self!capture($callbacks,'stdout',$!stdout_supplier),
$!stdout_descriptor_used
)) if $!stdout_supply;
)) if $!stdout_supplier;
@!promises.push(Promise.anyof(
self!capture($callbacks,'stderr',$!stderr_supply),
self!capture($callbacks,'stderr',$!stderr_supplier),
$!stderr_descriptor_used
)) if $!stderr_supply;
)) if $!stderr_supplier;
@!promises.push(
self!capture($callbacks,'merge',$!merge_supply)
) if $!merge_supply;
self!capture($callbacks,'merge',$!merge_supplier)
) if $!merge_supplier;

nqp::bindkey($callbacks, 'buf_type', buf8.new);
nqp::bindkey($callbacks, 'write', True) if $.w;
Expand All @@ -346,7 +364,7 @@ my class Proc::Async {
$callbacks,
);
$!handle_available_promise.keep(True);
nqp::permit($!process_handle, 0, -1) if $!merge_supply;
nqp::permit($!process_handle, 0, -1) if $!merge_supplier;
Promise.allof( $!exit_promise, @!promises ).then({
.close for @!close-after-exit;
$!exit_promise.status == Broken
Expand Down Expand Up @@ -412,15 +430,24 @@ my class Proc::Async {
proto method kill(|) {*}
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Might be worth consolidating all the duplicated logic (including some existing) into the proto:

proto method kill(|) {
    X::Proc::Async::MustBeStarted.new(:method<kill>, proc => self).throw if !$!started;
    $!stdout_supplier.done() if $!stdout_supplier;
    $!stderr_supplier.done() if $!stderr_supplier;
    $!merge_supplier.done()  if $!merge_supplier;
    {*}
}

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I had left the duplicate logic because of the comment a few lines above stating that it significantly impacted performance. That comment is quite old though so things could be different now with the improvements made to the optimizer and spesh.

multi method kill(Proc::Async:D: Signal:D \signal = SIGHUP) {
X::Proc::Async::MustBeStarted.new(:method<kill>, proc => self).throw if !$!started;
$!stdout_supplier.done() if $!stdout_supplier;
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Will users expect &done or &quit callbacks to be fired? I'm not really sure.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Currently, the killprocasync op invoked in the method call always cancels the work on the event queue (in MoarVM at least). However, not knowing that, it isn't unreasonable that the user might expect that the process could recover from the signal.

I was originally approaching it from the standpoint that the user is going in expecting the termination of the process, but I'm not so sure now. Might be too late to change the behavior though?

$!stderr_supplier.done() if $!stderr_supplier;
$!merge_supplier.done() if $!merge_supplier;
nqp::killprocasync($!process_handle, signal.value)
}
multi method kill(Proc::Async:D: Int:D \signal) {
X::Proc::Async::MustBeStarted.new(:method<kill>, proc => self).throw if !$!started;
$!stdout_supplier.done() if $!stdout_supplier;
$!stderr_supplier.done() if $!stderr_supplier;
$!merge_supplier.done() if $!merge_supplier;
nqp::killprocasync($!process_handle, signal)
}

multi method kill(Proc::Async:D: Str:D \signal) {
X::Proc::Async::MustBeStarted.new(:method<kill>, proc => self).throw if !$!started;
$!stdout_supplier.done() if $!stdout_supplier;
$!stderr_supplier.done() if $!stderr_supplier;
$!merge_supplier.done() if $!merge_supplier;
nqp::killprocasync($!process_handle, $*KERNEL.signal: signal)
}
}
Expand Down