lib/IPC/Semaphore/Concurrency.pm

   1 package IPC::Semaphore::Concurrency;
   2
   3 use 5.008008;
   4 use strict;
   5 use warnings;
   6
   7 use Carp;
   8 use POSIX qw(O_WRONLY O_CREAT O_NONBLOCK O_NOCTTY);
   9 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);
  10 use IPC::Semaphore;
  11
  12 require Exporter;
  13 our @ISA = qw(Exporter);
  14 our %EXPORT_TAGS = ( 'all' => [ qw() ] );
  15 our @EXPORT_OK = ( @{ $EXPORT_TAGS{'all'} } );
  16 our @EXPORT = qw();
  17
  18 our $VERSION = '0.02';
  19
  20 sub new {
  21         my $class = shift;
  22
  23         my %args;
  24         if (@_ == 1) {
  25                 # Only one required argument
  26                 $args{'path'} = shift;
  27         } else {
  28                 %args = @_;
  29         }
  30
  31         if (!exists($args{'path'})) {
  32                 carp "Must supply a path!"; #TODO: Allow private semaphores
  33                 return undef;
  34         }
  35         # Set defaults
  36         $args{'project'} = 0 if (!exists($args{'project'}));
  37         $args{'count'} = 1 if (!exists($args{'count'}));
  38         $args{'value'} = 1 if (!exists($args{'value'})); # TODO: allow array (one value per semaphore)
  39         $args{'touch'} = 1 if (!exists($args{'touch'}));
  40
  41         my $self = bless {}, $class;
  42         $self->{'_args'} = { %args };
  43
  44         $self->_touch($self->{'_args'}->{'path'}) if (!-e $self->{'_args'}->{'path'} || $self->{'_args'}->{'touch'}) or return undef;
  45         $self->{'_args'}->{'key'} = $self->_ftok() or return undef;
  46
  47         $self->{'_args'}->{'sem'} = $self->_create($self->key()) or return undef;
  48
  49         return $self;
  50 }
  51
  52 # Internal functions
  53 sub _touch {
  54         # Create and/or touch the path, returns false if there's an error
  55         my $self = shift;
  56         my $path = shift;
  57         sysopen(my $fh, $path, O_WRONLY|O_CREAT|O_NONBLOCK|O_NOCTTY) or carp "Can't create $path: $!" and return 0;
  58         utime(undef, undef, $path) if ($self->{'_args'}->{'touch'});
  59         close $fh or carp "Can't close $path: $!" and return 0;
  60         return 1;
  61 }
  62
  63 sub _ftok {
  64         # Create an IPC key, returns result of ftok()
  65         my $self = shift;
  66         return ftok($self->{'_args'}->{'path'}, $self->{'_args'}->{'project'}) or carp "Can't create semaphore key: $!" and return undef;
  67 }
  68
  69 sub _create {
  70         # Create the semaphore and assign it its initial value
  71         my $self = shift;
  72         my $key = shift;
  73         # Presubably the semaphore exists already, so try using it right away
  74         my $sem = IPC::Semaphore->new($key, 0, 0);
  75         if (!defined($sem)) {
  76                 # Creatie a new semaphore...
  77                 $sem = IPC::Semaphore->new($key, $self->{'_args'}->{'count'}, IPC_CREAT|IPC_EXCL|S_IRUSR|S_IWUSR|S_IRGRP|S_IWGRP|S_IROTH|S_IWOTH);
  78                 if (!defined($sem)) {
  79                         # Make sure another process did not create it in our back
  80                         $sem = IPC::Semaphore->new($key, 0, 0) or carp "Semaphore creation failed!\n";
  81                 } else {
  82                         # If we created the semaphore now we assign its initial value
  83                         for (my $i=0; $i<$self->{'_args'}->{'count'}; $i++) { # TODO: Support array - see above
  84                                 $sem->op($i, $self->{'_args'}->{'value'}, 0);
  85                         }
  86                 }
  87         }
  88         # Return whatever last semget call got us
  89         return $sem;
  90 }
  91
  92 # External API
  93
  94 sub getall {
  95         my $self = shift;
  96         return $self->{'_args'}->{'sem'}->getall();
  97 }
  98
  99 sub getval {
 100         my $self = shift;
 101         my $nsem = shift or 0;
 102         return $self->{'_args'}->{'sem'}->getval($nsem);
 103 }
 104
 105 sub getncnt {
 106         my $self = shift;
 107         my $nsem = shift or 0;
 108         return $self->{'_args'}->{'sem'}->getncnt($nsem);
 109 }
 110
 111 sub setall {
 112         my $self = shift;
 113         return $self->{'_args'}->{'sem'}->setall(@_);
 114 }
 115
 116 sub setval {
 117         my $self = shift;
 118         my ($nsem, $val) = @_;
 119         return $self->{'_args'}->{'sem'}->setval($nsem, $val);
 120 }
 121
 122 sub stat {
 123         my $self = shift;
 124         return $self->{'_args'}->{'sem'}->stat();
 125 }
 126
 127 sub id {
 128         my $self = shift;
 129         return $self->{'_args'}->{'sem'}->id();
 130 }
 131
 132 sub key {
 133         my $self = shift;
 134         return $self->{'_args'}->{'key'};
 135 }
 136
 137 sub acquire {
 138         my $self = shift;
 139
 140         my %args;
 141         if (@_ >= 1 && $_[0] =~ /^\d+$/) {
 142                 # Positional arguments
 143                 ($args{'sem'}, $args{'wait'}, $args{'max'}, $args{'undo'}) = @_;
 144         } else {
 145                 %args = @_;
 146         }
 147         # Defaults
 148         $args{'sem'} =  0 if (!defined($args{'sem'}));
 149         $args{'wait'} =  0 if (!defined($args{'wait'}));
 150         $args{'max'} = -1 if (!defined($args{'max'}));
 151         $args{'undo'} = 1 if (!defined($args{'undo'}));
 152
 153         my $sem = $self->{'_args'}->{'sem'};
 154         my $flags = IPC_NOWAIT;
 155         $flags |= SEM_UNDO if ($args{'undo'});
 156
 157         my ($ret, $ncnt);
 158         # Get blocked process count here to retain Errno (thus $!) after the first semop call.
 159         $ncnt = $self->getncnt($args{'sem'}) if ($args{'wait'});
 160
 161         if (($ret = $sem->op($args{'sem'}, -1, $flags))) {
 162                 return $ret;
 163         } elsif ($args{'wait'}) {
 164                 return $ret if ($args{'max'} >= 0 && $ncnt >= $args{'max'});
 165                 # Remove NOWAIT and block
 166                 $flags ^= IPC_NOWAIT;
 167                 return $sem->op($args{'sem'}, -1, $flags);
 168         }
 169         return $ret;
 170 }
 171
 172 sub release {
 173         my $self = shift;
 174         my $number = shift || 0;
 175         return $self->{'_args'}->{'sem'}->op($number, 1, 0);
 176 }
 177
 178 sub remove {
 179         my $self = shift;
 180         return $self->{'_args'}->{'sem'}->remove();
 181 }
 182
 183 1;
 184 __END__
 185
 186 =head1 NAME
 187
 188 IPC::Semaphore::Concurrency - Concurrency guard using semaphores
 189
 190 =head1 SYNOPSIS
 191
 192     use IPC::Semaphore::Concurrency;
 193
 194     my $c = IPC::Semaphore::Concurrency->new('/tmp/sem_file');
 195
 196     if ($c->acquire()) {
 197         print "Do work\n";
 198     } else {
 199         print "Pass our turn\n";
 200     }
 201
 202
 203     my $c = IPC::Semaphore::Concurrency->new(
 204         path  => '/tmp/sem_file',
 205         count => 2,
 206         value => $sem_max,
 207         );
 208
 209     if ($c->acquire(0, 1, 0)) {
 210         print "Do work\n";
 211     } else {
 212         print "Error: Another process is already locked\n";
 213     }
 214
 215     if ($c->acquire(1)) {
 216         print "Do other work\n";
 217     }
 218
 219 =head1 DESCRIPTION
 220
 221 This module allows you to limit concurrency of specific portions of your
 222 code. It can be used to limit resource usage or to give exclusive access to
 223 specific resources.
 224
 225 This module is similar in functionality to IPC::Concurrency with the main
 226 differences being that is uses SysV Semaphores, and allow queuing up
 227 processes while others hold the semaphore. There are other difference which
 228 gives more flexibility in some cases.
 229
 230 Generally, errors messages on failures can be retrieved with $!.
 231
 232 =head2 EXPORTS
 233
 234 None for now (could change before first Beta)
 235
 236 =head1 CONSTRUCTOR
 237
 238     IPC::Semaphore::Concurrency->new( $path );
 239
 240     IPC::Semaphore::Concurrency->new(
 241         path    => $path
 242         project => $proj_id
 243         count   => $sem_count
 244         value   => $sem_value
 245         touch   => $touch_path
 246         );
 247
 248 =over 4
 249
 250 =item path
 251
 252 The path to combine with the project id for creating the semaphore key.
 253 This file is only used for the inode and device numbers. Will be created
 254 if missing.
 255
 256 =item project
 257
 258 The project_id used for generating the key. If nothing else, the
 259 semaphore value can be used as changing the count will force generating a
 260 new semaphore. Defaults to 0.
 261
 262 =item count
 263
 264 Number of semaphores to create. Default is 1.
 265
 266 =item value
 267
 268 Value assigned to the semaphore at creation time. Default is 1.
 269
 270 =item touch
 271
 272 If true, tough the path when creating the semaphore. This can be used to
 273 ensure a file in /tmp do not get removed because it is too old.
 274
 275 =back
 276
 277 =head1 FUNCTIONS
 278
 279 =head2 getall
 280
 281 =head2 getval
 282
 283 =head2 getncnt
 284
 285 =head2 id
 286
 287 =head2 setall
 288
 289 =head2 setval
 290
 291 =head2 stat
 292
 293 =head2 remove
 294
 295 These functions are wrapper of the same functions in IPC::Semaphore.
 296
 297 For getval and getncnt, if no argument is given the default is 0.
 298
 299 =head2 key
 300
 301     $c->key();
 302
 303 Return the key used to create the semaphore.
 304
 305 =head2 acquire
 306
 307     $c->acquire();
 308
 309     $c->acquire($sem_number, $wait, $max, $undo);
 310
 311     $c->acquire(
 312         sem  => $sem_number,
 313         wait => $wait,
 314         max  => $max,
 315         undo => $undo,
 316         );
 317
 318 Acquire a semaphore lock. Return true if the lock was acquired.
 319
 320 =over 4
 321
 322 =item sem
 323
 324 The semaphore number to get. Defaults to 0.
 325
 326 =item wait
 327
 328 If true, block on semaphore acquisition.
 329
 330 =item max
 331
 332 If C<wait> is true, don't block if C<max> processes or more are waiting
 333 for the semaphore. Defaults to -1 (unlimited).
 334
 335 You may want to set it to some decent value if blocking on the semaphore
 336 to ensure processes don't add up infinitely.
 337
 338 =item undo
 339
 340 If defined and false, the semaphore won't be released automatically when
 341 process exits. You must release manually and B<only once> the semaphore
 342 with C<< $c->release() >>. See C<release> for important information before using
 343 this!
 344
 345 Use with caution as you can block semaphore slots if the process crash or
 346 gets killed.
 347
 348 =back
 349
 350 =head2 release
 351
 352     $c->release();
 353
 354     $c->release($sem_number);
 355
 356 B<WARNING: Use this at your own risks and only after understanding the
 357 implications below!>
 358
 359 This function is useful only if you turn off the C<undo> option in
 360 C<acquire> function and B<must be used with it.> It merely increment the
 361 semaphore by one.
 362
 363 In its usual use case, IPC::Semaphore::Concurrency is locked once and
 364 until the process exits. This function allow you to control yourself the
 365 release of the lock, but you must understand what releasing a semaphore
 366 actually means. Semaphores are merely counters and every time you
 367 C<acquire> them you merely decrease the value - the locking happens once
 368 the counter reaches 0.
 369
 370 This means if you C<release> more than once, you will effectively
 371 increase the semaphore value and next time the semaphore is used it will
 372 require as many C<acquire> to lock or fail locking. B<This includes the
 373 implicit increase when the process exits when you don't set C<undo> to
 374 false in C<acquire>!>. This means if you use C<release> without C<undo>
 375 set to false, you will raise the value again at every process exit and your
 376 semaphore won't lock things anymore!
 377
 378 =head1 TODO
 379
 380 =head3 Allow setting semaphore permissions, default to 600
 381
 382 =head3 Allow private semaphores
 383
 384 =head3 Allow passing an array of values
 385
 386 =head1 BUGS
 387
 388 semop(3) and semop(3p) man pages both indicate that C<errno> should be set to
 389 C<EAGAIN> if the call would block and C<IPC_NOWAIT> is used, yet in my tests
 390 under Linux C<errno> was set to C<EWOULDBLOCK>. See C<example.pl> and
 391 C<example2.pl> for examples of paranoiac error checking. YMMV.
 392
 393 Please report bugs to C<tguyot@gmail.com>.
 394
 395 =head1 SEE ALSO
 396
 397 L<IPC::Semaphore> - The module this is based on.
 398
 399 The code repository is mirrored on
 400 L<http://repo.or.cz/w/IPC-Semaphore-Concurrency.git>
 401
 402 CLI tools for controlling semaphores:
 403
 404 ipcs(1), especially ipcs -s for listing all semaphores
 405
 406 ipcrm(1), for removing semaphores by ID (-s) or KEY (-S)
 407
 408 =head1 AUTHOR
 409
 410 Thomas Guyot-Sionnest <tguyot@gmail.com>
 411
 412 =head1 COPYRIGHT AND LICENSE
 413
 414 Copyright (C) 2009 Thomas Guyot-Sionnest <tguyot@gmail.com>
 415
 416 This library is free software; you can redistribute it and/or modify
 417 it under the same terms as Perl itself, either Perl version 5.8.8 or,
 418 at your option, any later version of Perl 5 you may have available.
 419
 420 =cut