interprocess communication in perl

spawn.pl

#!/usr/bin/env perl

use strict;
use warnings;
use utf8;
use 5.014;
use experimental qw/smartmatch/;

use Carp qw/croak/;
use Data::Dumper;
use JSON ();
use POSIX qw/:sys_wait_h :signal_h/;

my $PARENT_PID = $$;
my $JSON = JSON->new->utf8;

my %PIDS;

pipe my ( $parent_reader, $child_writer ) or croak "Cannot create pipe: $!";
$_->autoflush(1) for ( $parent_reader, $child_writer );

my $DEFAULT_READER = $parent_reader;
my $DEFAULT_WRITER;

$SIG{CHLD} = \&reap_child;

sub reap_child {
    warn "get CHLD signal\n";
    while ( ( my $pid = waitpid -1, WNOHANG ) > 0 ) {
        warn "Process $pid exited\n";
        delete $PIDS{ $pid };
    }
    $SIG{CHLD} = \&reap_child;
}

sub kill_all {
    kill TERM => keys %PIDS;
    sleep 1 while scalar %PIDS;
}

sub encode {
    my ( $perl_scalar ) = @_;
    return $JSON->encode( $perl_scalar );
}

sub decode {
    my ( $text ) = @_;
    return $JSON->decode( $text );
}

sub receive_from {
    my ( $reader ) = @_;

    return if not defined( my $line = <$reader> );

    chomp( $line );
    return decode( $line );
}

sub receive {
    return receive_from( $DEFAULT_READER );
}

sub transmit_to {
    my ( $writer, $perl_scalar ) = @_;
    printf { $writer } "%s\n", encode( $perl_scalar );
}

sub transmit {
    my ( $pid, $perl_scalar);
    if ( scalar @_ > 1 ) {
        ( $pid, $perl_scalar ) = @_;
    } else {
        ( $perl_scalar ) = @_;
        $pid = $PARENT_PID;
    }
    my $writer = ( $pid == $PARENT_PID ) ? $child_writer : $PIDS{$pid}{pipe};
    transmit_to( $writer, $perl_scalar );
}

sub naked_spawn {
    my ( $code_ref, $parent_writer, $child_reader ) = @_;

    my $signals = POSIX::SigSet->new( SIGCHLD, SIGHUP, SIGINT, SIGTERM );
    sigprocmask( SIG_BLOCK, $signals ) or croak "Cannot block signals: $!";

    my $pid = fork;
    croak "Cannot fork: $!" if not defined $pid;

    if ( $pid ) {
        sigprocmask( SIG_UNBLOCK, $signals ) or croak "Cannot unblock signals: $!";
        close $child_reader;
    } else {
        $SIG{CHLD} = $SIG{HUP} = $SIG{INT} = $SIG{TERM} = 'DEFAULT';
        sigprocmask( SIG_UNBLOCK, $signals ) or croak "Cannot unblock signals: $!";

        undef %PIDS;
        $DEFAULT_READER = $child_reader;

        close $_ for ( $parent_reader, $parent_writer );

        $code_ref->( $child_reader, $child_writer );

        close $_ for ( $child_reader, $child_writer );
        exit 0;
    }


    return $pid;
}

sub spawn {
    my ( $code_ref ) = @_;

    pipe my ( $child_reader, $parent_writer ) or croak "Cannot create pipe: $!";
    $_->autoflush(1) for ( $child_reader, $parent_writer );

    my $pid = naked_spawn( $code_ref, $parent_writer, $child_reader );

    $PIDS{$pid}{pipe} = $parent_writer;

    return $pid;
}

sub loop {
    local @_ = @_;
    my ( $reader, $writer ) = @_;

    return if not defined( my $info = receive() );

    for ( $info->{action} ) {
        when ('add') {
            transmit( { content => "$$ Add: $info->{content}" } );
            goto &loop;
        }
        when ('delete') {
            transmit( { content => "$$ Delete: $info->{content}" } );
            goto &loop;
        }
        when ( 'exit' ) {
            transmit( { content => "$$ Exit" } );
        }
        default {
            transmit( { content => "$$ Unknown action: $info->{action}" } );
            goto &loop;
        }
    }
}

my @children = map { spawn(\&loop) } 0 .. 3;

transmit( $children[0], { action => 'add', content => 'hello' } );
transmit( $children[0], { action => 'add', content => 'hello again' } );
transmit( $children[1], { action => 'delete', content => 'world' } );
transmit( $children[2], { action => 'update', content => '!' } );
transmit( $children[3], { action => 'exit' } );

say "parent pid: $PARENT_PID";
my $DONE = 0;
$SIG{INT} = $SIG{TERM} = sub { $DONE = 1; };

while ( not $DONE ) {
    next if not my $info = eval {
        $SIG{INT} = $SIG{TERM} = sub { $DONE = 1; die; };
        receive();
    };
    print Dumper $info;
}

kill_all();
say "Normal exited";