File Coverage

lib/AnyEvent/ForkObject.pm

Criterion	Covered	Total	%
statement	31	33	93.9
branch			n/a
condition			n/a
subroutine	11	11	100.0
pod			n/a
total	42	44	95.4

package AnyEvent::ForkObject;

use 5.010001;
use strict;
use warnings;

use Carp;
use AnyEvent;
use AnyEvent::Util;
use AnyEvent::Handle;
use Scalar::Util qw(weaken blessed reftype);
use POSIX;
use IO::Handle;
use AnyEvent::Serialize qw(:all);
use AnyEvent::Tools qw(mutex);
use Devel::GlobalDestruction;


our $VERSION = '0.09';

sub new
{
    my ($class) = @_;

    my $self = bless { } => ref($class) || $class;
    my ($s1, $s2) = portable_socketpair;

    if ($self->{pid} = fork) {
        # parent
        $self->{mutex} = mutex;
        close $s2;
        fh_nonblocking $s1, 1;
        {
            weaken(my $self = $self);
            $self->{handle} = new AnyEvent::Handle
                fh => $s1,
                on_error => sub {
                    return unless $self;
                    return if $self->{destroyed};
                    delete $self->{handle};
                    $self->{fatal} = $!;
                    $self->{cb}(fatal => $self->{fatal}) if $self->{cb};
                };
        }
    } elsif (defined $self->{pid}) {
        # child
        close $s1;
        $self->{socket} = $s2;
        $self->{object} = {};
        $self->{no} = 0;
        $self->_start_server;
    } else {
        die $!;
    }

    return $self;
}

sub do :method
{
    my ($self, %opts) = @_;
    my $method = $opts{method} || 'new';
    my $invocant = $opts{module} || $opts{_invocant};
    my $cb = $opts{cb} || sub {  };
    my $args = $opts{args} || [];
    my $wantarray = $opts{wantarray};
    my $require = $opts{require};
    $wantarray = 0 unless exists $opts{wantarray};

    weaken $self;
    $self->{mutex}->lock(sub {
        my ($guard) = @_;
        return unless $self;
        return if $self->{destroyed};

        $self->{cb} = $cb;

        unless ($self->{handle}) {
            $cb->(fatal => 'Child process was destroyed');
            undef $guard;
            return;
        }

        if ($self->{fatal}) {
            $cb->(fatal => $self->{fatal});
            delete $self->{cb};
            undef $guard;
            return;
        }

        serialize {
                $require ? (r => $require) : (
                    i   => $invocant,
                    m   => $method,
                    a   => $args,
                    wa  => $wantarray
                )
            } => sub {
                return unless $self;
                return if $self->{destroyed} or $self->{fatal};

                $self->{handle}->push_write("$_[0]\n");
                return unless $self;
                return if $self->{destroyed} or $self->{fatal};

                $self->{handle}->push_read(line => "\n", sub {
                    deserialize $_[1] => sub {
                        return unless $self;
                        return if $self->{destroyed} or $self->{fatal};

                        my ($o, $error, $tail) = @_;

                        if ($error) {
                            $cb->(fatal => $error);
                            delete $self->{cb};
                            undef $guard;
                            return;
                        }

                        my $status = shift @$o;
                        if ($status eq 'ok') {
                            for (@$o) {
                                if (exists $_->{obj}) {
                                    $_ = bless {
                                        no => "$_->{obj}",
                                        fo => \$self,
                                    } => 'AnyEvent::ForkObject::OneObject';
                                    next;
                                }

                                $_ = $_->{res};
                            }
                            $cb->(ok => @$o);
                        } else {
                            $cb->($status => @$o);
                        }
                        delete $self->{cb};
                        undef $guard;
                    };
                    return;
                });

                return;
            };
    });

    return;
}


sub DESTROY
{
    my ($self) = @_;
    $self->{destroyed} = 1;
    $self->{handle}->push_write("'bye'\n") if $self->{handle};
    delete $self->{handle};

    return if in_global_destruction;

    # kill zombies
    my $cw;
    $cw = AE::child $self->{pid} => sub {
        my ($pid, $code) = @_;
        undef $cw;
    };
}

sub _start_server
{
    my ($self) = @_;
    croak "Something wrong" if $self->{pid};
    my $err_code = 0;

    require Data::StreamSerializer;

    my $socket = $self->{socket};
    $socket->autoflush(1);
    while(<$socket>) {
        my $response;
        next unless /\S/;
        my $cmd = eval $_;
        if ($@) {
            $err_code = 1;
            last;
        }

        unless (ref $cmd) {
            if ($cmd eq 'bye') {
                undef $_ for values %{ $self->{object} };
                delete $self->{object};
                last;
            }

            eval $cmd;

            if ($@) {
                $response = [ die => $@ ];
                goto RESPONSE;
            }

            $response = [ 'ok' ];
            goto RESPONSE;
        }

        # require
        if ($cmd->{r}) {
            eval "require $cmd->{r}";
            if ($@) {
                $response = [ die => $@ ];
                goto RESPONSE;
            }

            $response = [ 'ok' ];
            goto RESPONSE;
        }


        my ($invocant, $method, $args, $wantarray) = @$cmd{qw(i m a wa)};
        if ($invocant =~ /^\d+$/) {
            if ($method eq 'DESTROY') {
                delete $self->{object}{$invocant};
                $response = [ 'ok' ];
                goto RESPONSE;
            } else {
                $invocant = $self->{object}{$invocant}
            }
        }

        my @o;

        if ($method eq 'fo_attr') {
            unless (ref $invocant) {
                $response = [ die => 'fo_attr should be called as method' ];
                goto RESPONSE;
            }

            if ('ARRAY' eq reftype $invocant) {
                $invocant->[ $args->[0] ] = $args->[1] if @$args > 1;
                $o[0] = $invocant->[ $args->[0] ];
            } elsif ('HASH' eq reftype $invocant) {
                $invocant->{ $args->[0] } = $args->[1] if @$args > 1;
                $o[0] = $invocant->{ $args->[0] };
            } else {
                $response = [
                    die => "fo_attr can't access on blessed " .
                        reftype $invocant
                ];
                goto RESPONSE;
            }

        } else {
            if ($wantarray) {
                @o = eval { $invocant -> $method ( @$args ) };
            } elsif (defined $wantarray) {
                $o[0] = eval { $invocant -> $method ( @$args ) };
            } else {
                eval { $invocant -> $method ( @$args ) };
            }

            if ($@) {
                $response = [ die => $@ ];
                goto RESPONSE;
            }
        }

        for (@o) {
            if (ref $_ and blessed $_) {
                my $no = ++$self->{no};
                $self->{object}{$no} = $_;

                $_ = { obj => $no };
                next;
            }

            $_ = { res => $_ };
        }

        $response = [ ok => @o ];

        RESPONSE:
            my $sr = new Data::StreamSerializer($response);
            while(defined(my $part = $sr->next)) {
                print $socket $part;
            }
            print $socket "\n";
    }

    # destroy internal objects
    delete $self->{object};

    # we don't want to call any other destructors
    POSIX::_exit($err_code);
}

package AnyEvent::ForkObject::OneObject;
use Carp;
use Scalar::Util qw(blessed);
use Devel::GlobalDestruction;

sub AUTOLOAD
{
    our $AUTOLOAD;
    my ($foo) = $AUTOLOAD =~ /([^:]+)$/;

    my ($self, @args) = @_;
    my $cb = pop @args;
    my $wantarray = 0;
    if ('CODE' ne ref $cb) {
        $wantarray = $cb;
        $cb = pop @args;
    }
    croak "Callback is required" unless 'CODE' eq ref $cb;

    my $fo = $self->{fo};

    unless ($$fo) {
        $cb->(fatal => 'Child process was already destroyed');
        return;
    }

    $$fo -> do(
        _invocant => $self->{no},
        method    => $foo,
        args      => \@args,
        cb        => $cb,
        wantarray => $wantarray
    );
    return;
}

sub DESTROY
{
    # You can call DESTROY by hand
    my ($self, $cb) = @_;
    return if in_global_destruction;
    $cb ||= sub {  };
    my $fo = $self->{fo};
    unless (blessed $$fo) {
        $cb->(fatal => 'Child process was already destroyed');
        return;
    }

    $$fo -> do(
        _invocant   => $self->{no},
        method      => 'DESTROY',
        cb          => $cb,
        wantarray   => undef,
    );
    return;
}

1;
__END__