File Coverage

blib/lib/IPC/Semaphore/SmokeSignals.pm

Criterion	Covered	Total	%
statement	164	182	90.1
branch	54	90	60.0
condition	30	57	52.6
subroutine	30	32	93.7
pod	8	8	100.0
total	286	369	77.5

package IPC::Semaphore::SmokeSignals;
use strict;

use vars qw< $VERSION @EXPORT_OK >;
BEGIN {
    $VERSION = 0.001_002;
    @EXPORT_OK = qw< LightUp JoinUp MeetUp >;
    require IO::Handle;
    require Exporter;
    *import = \&Exporter::import;
    if(  eval { require bytes; 1 }  ) {
        bytes->import();
    }
}
use Time::HiRes qw< sleep >;
use Errno       qw< EAGAIN EWOULDBLOCK >;
use Fcntl       qw<
    O_WRONLY    O_RDONLY    O_NONBLOCK
    LOCK_EX     LOCK_NB     LOCK_UN
>;

sub _SMOKE { 0 }    # End to pull from.
sub _STOKE { 1 }    # The lit end.
sub _BYTES { 2 }    # Tokin' length.
sub _PUFFS { 3 }    # How many tokins; how many tokers at once (if I lit it).
sub _OWNER { 4 }    # PID of process that created this pipe.


sub LightUp {   # Set up a new pipe.
    return __PACKAGE__->Ignite( @_ );
}

sub JoinUp {    # Just use an existing pipe.
    return __PACKAGE__->JoinIn( @_ );
}

sub MeetUp {    # When you are not sure who should light the pipe.
    return __PACKAGE__->Meet( @_ );
}


sub _New {
    my( $class, $bytes, $path, $perm, $nowait ) = @_;

    my $smoke = IO::Handle->new();
    my $stoke = IO::Handle->new();
    if( ! $path ) {
        pipe( $smoke, $stoke )
            or  _croak( "Can't ignite pipe: $!\n" );
    } else {
        if( $perm && ! -e $path ) {
            warn "WARNING: Having to create FIFO: $path\n"
                if  $nowait;
            require POSIX;
            POSIX->import('mkfifo');    # In case import() says 'unsupported'.
            mkfifo( $path, $perm )
                or  _croak( "Can't create FIFO ($path): $!\n" );
        }
        my $extra = $perm || $nowait ? O_NONBLOCK() : 0;
        sysopen $smoke, $path, O_RDONLY()|$extra, $perm
            or  _croak( "Can't read pipe path ($path): $!\n" );
        _croak( "Path ($path) is not a FIFO (named pipe)\n" )
            if  ! -p $smoke;
        sysopen $stoke, $path, O_WRONLY()
            or  _croak( "Can't write pipe path ($path): $!\n" );
    }
    binmode $smoke;
    binmode $stoke;

    my $me = bless [], ref $class || $class;
    $me->[_SMOKE] = $smoke;
    $me->[_STOKE] = $stoke;
    $me->[_BYTES] = $bytes;

    return $me;
}


sub JoinIn {    # Use an already set-up pipe.
    my( $class, $bytes, $path ) = @_;
    my $me = $class->_New( $bytes, $path, 0 );
    return $me;
}


sub Ignite {    # Set up a new pipe.
    my( $class, $fuel, $path, $perm ) = @_;
    $perm ||= 0666
        if  $path;

    ( $fuel, my $bytes ) = $class->_PickTheMix( $fuel );

    my $me = $class->_New( $bytes, $path, $perm );

    $me->_Roll( $fuel );

    return $me;
}


sub _PickTheMix {
    my( $class, $fuel ) = @_;
    $fuel ||= 1;
    my $bytes;
    if( ref $fuel ) {
        _croak( "You brought nothing to smoke!\n" )
            if  ! @$fuel;
        $bytes = length $fuel->[0];
    } else {
        _croak( "Specify what to smoke or how much, not '$fuel'.\n" )
            if  $fuel !~ /^[1-9][0-9]*$/;
        $bytes = length $fuel;
    }
    return( $fuel, $bytes );
}


sub Meet {      # When you are not sure who should light the pipe.
    my( $class, $fuel, $path, $perm ) = @_;

    ( $fuel, my $bytes ) = $class->_PickTheMix( $fuel );

    my $me = $class->_New( $bytes, $path, $perm, 'nowait' );

    # See if somebody already lit the pipe:
    if( flock( $me->[_SMOKE], LOCK_EX() | LOCK_NB() ) ) {
        my $puff = $me->_Bogart('impatient');
        if( defined $puff ) {
            # Already lit, so return the magic smoke:
            $me->_Stoke( $puff );
        } else {
            # I got here first!  Light it up!
            $me->_Roll( $fuel );
        }
        flock( $me->[_SMOKE], LOCK_UN() );
    }

    return $me;
}


sub _Roll {     # Put the fuel in.
    my( $me, $fuel ) = @_;
    $me->[_OWNER] = $$;

    my $stoke = $me->[_STOKE];
    $stoke->blocking( 0 );
    if( ! ref $fuel ) {
        $me->[_PUFFS] = 0 + $fuel;
        my $start = '0' x length $fuel;
        $start =~ s/0$/1/;
        for my $puff (  "$start" .. "$fuel"  ) {
            $me->_Stoke( $puff );
        }
    } else {
        $me->[_PUFFS] = 0 + @$fuel;
        for my $puff (  @$fuel  ) {
            $me->_Stoke( $puff );
            _croak( "You can't use a string of null bytes as a tokin'" )
                if  $puff !~ /[^\0]/;
        }
    }
    $stoke->blocking( 1 );
}


sub _MagicDragon {  # Every magic dragon needs a good name.
    return __PACKAGE__ . '::Puff';
}


sub Puff {          # Get a magic dragon so you won't forget to share.
    my( $me, $impatient ) = @_;
    if( ref $me->[_PUFFS] ) {
        return
            if  wantarray;
        _croak( "The pipe is going out.\n" );
    }
    return $me->_MagicDragon()->_Inhale( $me, $impatient );
}


sub _Bogart {       # Take a drag (skipping proper protocol).
    my( $me, $impatient, $nil ) = @_;
    my( $smoke ) = $me->[_SMOKE];
    $smoke->blocking( 0 )
        if  $impatient;
    my $puff;
    my $got_none = ! sysread( $smoke, $puff, $me->[_BYTES] );
    my $excuse = $!;
    $smoke->blocking( 1 )
        if  $impatient;
    return undef
        if  $impatient
        &&  $got_none
        &&  (   EAGAIN() == $excuse
            ||  EWOULDBLOCK() == $excuse )
    ;
    _croak( "Can't toke pipe: $!\n" )
        if  $got_none;
    if( ! $nil && $puff !~ /[^\0]/ ) {  # Pipe is being smothered.
        $me->_Stoke( $puff );
        $me->_Snuff();                  # Stop us from using it.
        return
            if  wantarray;
        _croak( "The pipe is going out.\n" );
    }
    return $puff;
}


sub _Stoke {        # Return some magic smoke (skipping proper protocol).
    my( $me, $puff ) = @_;
    my $stoke = $me->[_STOKE];
    my $bytes = $me->[_BYTES];
    if(  $bytes != length $puff  ) {
        _croak( "Tokin' ($puff) is ", length($puff), " bytes, not $bytes!" );
    }
    syswrite( $stoke, $puff )
        or  die "Can't stoke pipe (with '$puff'): $!\n";
}


# Returns undef if we aren't allowed to extinguish it.
# Returns 0 if pipe is now completely extinguished.
# Otherwise, returns number of outstanding tokins remaining.

sub Extinguish {    # Last call!
    my( $me, $impatient ) = @_;
    my $puffs = $me->[_PUFFS];

    my $left = undef;           # Returned if we didn't start the fire.
    $left = 0                   # Returned when it is all the way out...
        if  defined $puffs      # ...since we did start the fire.
        &&  $$ == ( $me->[_OWNER] || 0 );

    return $left                # We already put out at least ours.
        if  ref $puffs && ! @$puffs;

    if( defined $left ) {       # We brought it up so we can take it down:
        $left = $me->_Smother( $impatient );
        return $left            # Not all the way out yet.
            if  $left;
    }
    # Either all the way out or just extinguishing our access to it:
    $me->_Snuff();
    return $left;
}


sub _Snuff {
    my( $me ) = @_;
    for my $puffs ( $me->[_PUFFS] ) {
        return
            if  ref $puffs && ! @$puffs;
        $puffs = [];
    }
    close $me->[_STOKE];
    close $me->[_SMOKE];
}


sub _Smother {
    my( $me, $impatient ) = @_;
    my $puffs = $me->[_PUFFS];
    my $eop = "\0" x $me->[_BYTES]; # The End-Of-Pipe tokin'.
    my $eops;                       # How many EOPs in pipe?
    my $room;                       # How much room in pipe for EOP tokins?
    if( ! ref $puffs ) {            # Our first try at shutting down:
        $room = $eops = 0;          # Nothing drained, no EOPs sent.
    } else {
        ( $puffs, $room, $eops ) = @$puffs;
    }

    my $left;
    my $loops = 0;
    while( 0 < ( $left = $puffs + $eops ) ) {
        my $puff = $me->_Bogart(
            $impatient || ! $eops,      # Don't wait before injecting EOP.
            'nil' );
        if( ! defined $puff ) {         # Pipe empty:
            if( ! $room && ! $eops ) {  # "No room" but pipe empty:
                $me->_Stoke( $eop ); ++$eops;   # Risk just 1 EOP.
            }
        } elsif( $puff =~ /[^\0]/ ) {   # We eliminated another non-EOP tokin':
            --$puffs;
            ++$room if $room < 2;
            $loops = 0;                 # Don't sleep while reaping tokins.
        } else {
            --$eops;                    # Got an EOP back.
        }
        last                            # All puffed out!
            if  ! $puffs && ! $eops;
        # Don't inject EOPs if would just cause $impatient to never return:
        if(     ! $impatient        # If patient, then we might loop w/ sleep.
            ||  ! defined $puff     # We are about to return, so inject.
            ||  $puff =~ /[^\0]/    # We got a non-EOP tokin' so add more EOP.
        ) {
            while( $puffs && $eops < $room ) {
                $me->_Stoke( $eop ); ++$eops;
            }
        }
        if( $impatient && ! defined $puff ) {   # We had emptied the pipe:
            $me->[_PUFFS] = [$puffs,$room,$eops];
            return $left;                       # Report: Others need time.
        }
        sleep( 0.1 )                            # Don't do a tight CPU loop.
            if  2 < ++$loops;
    }
    return 0;
}


sub _croak {
    require Carp;
    Carp::croak( @_ );
}


our @CARP_NOT;

package IPC::Semaphore::SmokeSignals::Puff;
push @CARP_NOT, __PACKAGE__;

sub _Inhale {
    my( $class, $pipe, $impatient ) = @_;
    my( $puff ) = $pipe->_Bogart($impatient)
        or  return;
    $puff   or  return undef;
    return bless [ $pipe, $puff ], $class;
}

sub Sniff {
    my( $me ) = @_;
    return $me->[1];
}

sub Exhale {
    my( $me ) = @_;
    return
        if  ! @$me;
    my( $pipe, $puff ) = splice @$me;
    $pipe->_Stoke( $puff );
}

sub DESTROY {
    my( $me ) = @_;
    $me->Exhale();
}


1;
__END__