File Coverage

blib/lib/IPC/Semaphore/Concurrency.pm

Criterion	Covered	Total	%
statement	58	101	57.4
branch	18	56	32.1
condition	2	19	10.5
subroutine	13	22	59.0
pod	11	12	91.6
total	102	210	48.5

package IPC::Semaphore::Concurrency;

use 5.008008;
use strict;
use warnings;

use Carp;
use POSIX qw(O_WRONLY O_CREAT O_NONBLOCK O_NOCTTY);
use IPC::SysV qw(ftok IPC_NOWAIT IPC_CREAT IPC_EXCL S_IRUSR S_IWUSR S_IRGRP S_IWGRP S_IROTH S_IWOTH SEM_UNDO);
use IPC::Semaphore;

require Exporter;
our @ISA = qw(Exporter);
our %EXPORT_TAGS = ( 'all' => [ qw() ] );
our @EXPORT_OK = ( @{ $EXPORT_TAGS{'all'} } );
our @EXPORT = qw();

our $VERSION = '0.04';

sub new {
	my $class = shift;

	my %args;
	if (@_ == 1) {
		# Only one required argument
		$args{'path'} = shift;
	} else {
		%args = @_;
	}

	if (!exists($args{'path'})) {
		carp "Must supply a path!"; #TODO: Allow private semaphores
		return undef;
	}
	# Set defaults
	$args{'project'} = 0 if (!exists($args{'project'}));
	$args{'count'} = 1 if (!exists($args{'count'}));
	$args{'value'} = 1 if (!exists($args{'value'})); # TODO: allow array (one value per semaphore)
	$args{'touch'} = 1 if (!exists($args{'touch'}));

	my $self = bless {}, $class;
	$self->{'_args'} = { %args };

	$self->_touch($self->{'_args'}->{'path'}) if (!-e $self->{'_args'}->{'path'} || $self->{'_args'}->{'touch'}) or return undef;
	$self->{'_args'}->{'key'} = $self->_ftok() or return undef;

	$self->{'_args'}->{'sem'} = $self->_create($self->key()) or return undef;

	return $self;
}

# Internal functions
sub _touch {
	# Create and/or touch the path, returns false if there's an error
	my $self = shift;
	my $path = shift;
	sysopen(my $fh, $path, O_WRONLY|O_CREAT|O_NONBLOCK|O_NOCTTY) or carp "Can't create $path: $!" and return 0;
	utime(undef, undef, $path) if ($self->{'_args'}->{'touch'});
	close $fh or carp "Can't close $path: $!" and return 0;
	return 1;
}

sub _ftok {
	# Create an IPC key, returns result of ftok()
	my $self = shift;
	return ftok($self->{'_args'}->{'path'}, $self->{'_args'}->{'project'}) or carp "Can't create semaphore key: $!" and return undef;
}

sub _create {
	# Create the semaphore and assign it its initial value
	my $self = shift;
	my $key = shift;
	# Presubably the semaphore exists already, so try using it right away
	my $sem = IPC::Semaphore->new($key, 0, 0);
	if (!defined($sem)) {
		# Creatie a new semaphore...
		$sem = IPC::Semaphore->new($key, $self->{'_args'}->{'count'}, IPC_CREAT|IPC_EXCL|S_IRUSR|S_IWUSR|S_IRGRP|S_IWGRP|S_IROTH|S_IWOTH);
		if (!defined($sem)) {
			# Make sure another process did not create it in our back
			$sem = IPC::Semaphore->new($key, 0, 0) or carp "Semaphore creation failed!\n";
		} else {
			# If we created the semaphore now we assign its initial value
			for (my $i=0; $i<$self->{'_args'}->{'count'}; $i++) { # TODO: Support array - see above
				$sem->op($i, $self->{'_args'}->{'value'}, 0);
			}
		}
	}
	# Return whatever last semget call got us
	return $sem;
}

# External API

sub getall {
	my $self = shift;
	return $self->{'_args'}->{'sem'}->getall();
}

sub getval {
	my $self = shift;
	my $nsem = shift or 0;
	return $self->{'_args'}->{'sem'}->getval($nsem);
}

sub getncnt {
	my $self = shift;
	my $nsem = shift or 0;
	return $self->{'_args'}->{'sem'}->getncnt($nsem);
}

sub setall {
	my $self = shift;
	return $self->{'_args'}->{'sem'}->setall(@_);
}

sub setval {
	my $self = shift;
	my ($nsem, $val) = @_;
	return $self->{'_args'}->{'sem'}->setval($nsem, $val);
}

sub stat {
	my $self = shift;
	return $self->{'_args'}->{'sem'}->stat();
}

sub id {
	my $self = shift;
	return $self->{'_args'}->{'sem'}->id();
}

sub key {
	my $self = shift;
	return $self->{'_args'}->{'key'};
}

sub acquire {
	my $self = shift;

	my %args;
	if (@_ >= 1 && $_[0] =~ /^\d+$/) {
		# Positional arguments
		($args{'sem'}, $args{'wait'}, $args{'max'}, $args{'undo'}) = @_;
	} else {
		%args = @_;
	}
	# Defaults
	$args{'sem'} =  0 if (!defined($args{'sem'}));
	$args{'wait'} =  0 if (!defined($args{'wait'}));
	$args{'max'} = -1 if (!defined($args{'max'}));
	$args{'undo'} = 1 if (!defined($args{'undo'}));

	my $sem = $self->{'_args'}->{'sem'};
	my $flags = IPC_NOWAIT;
	$flags |= SEM_UNDO if ($args{'undo'});

	my ($ret, $ncnt);
	# Get blocked process count here to retain Errno (thus $!) after the first semop call.
	$ncnt = $self->getncnt($args{'sem'}) if ($args{'wait'});

	if (($ret = $sem->op($args{'sem'}, -1, $flags))) {
		return $ret;
	} elsif ($args{'wait'}) {
		return $ret if ($args{'max'} >= 0 && $ncnt >= $args{'max'});
		# Remove NOWAIT and block
		$flags ^= IPC_NOWAIT;
		return $sem->op($args{'sem'}, -1, $flags);
	}
	return $ret;
}

sub release {
	my $self = shift;
	my $number = shift || 0;
	return $self->{'_args'}->{'sem'}->op($number, 1, 0);
}

sub remove {
	my $self = shift;
	return $self->{'_args'}->{'sem'}->remove();
}

1;
__END__