File Coverage

blib/lib/Parallel/Scoreboard.pm

Criterion	Covered	Total	%
statement	81	87	93.1
branch	20	36	55.5
condition	4	12	33.3
subroutine	16	18	88.8
pod	4	4	100.0
total	125	157	79.6

package Parallel::Scoreboard;

use strict;
use warnings;

use Digest::MD5 qw(md5);
use Fcntl qw(:flock);
use IO::Handle;
use POSIX qw(:fcntl_h);
use File::Path;

our $VERSION = '0.09';

use Class::Accessor::Lite (
    ro => [ qw(base_dir worker_id) ],
);

sub new {
    my $klass = shift;
    my %args = @_;
    die "mandatory parameter:base_dir is missing"
        unless $args{base_dir};
    # create base_dir if necessary
    if (! -e $args{base_dir}) {
	mkpath $args{base_dir}
            or -e $args{base_dir} or die "failed to create directory:$args{base_dir}:$!";
    }
    # build object
    my $self = bless {
        worker_id => sub { $$ },
        %args,
    }, $klass;
    # remove my status file, just in case
    unlink $self->_build_filename();

    return $self;
}

sub DESTROY {
    my $self = shift;
    # if file is open, close and unlink
    if ($self->{fh}) {
        close $self->{fh};
        # during global destruction we may already have lost this
	unlink $self->_build_filename() if ($self->{base_dir});
    }
}

sub update {
    my ($self, $status) = @_;
    # open file at the first invocation (tmpfn => lock => rename)
    my $id = $self->worker_id->();
    if ($self->{fh} && $self->{id_for_fh} ne $id) {
        # fork? close but do not unlock
        close $self->{fh};
        undef $self->{fh};
    }
    unless ($self->{fh}) {
        my $fn = $self->_build_filename();
        open my $fh, '>', "$fn.tmp"
            or die "failed to open file:$fn.tmp:$!";
        autoflush $fh 1;
        flock $fh, LOCK_EX
            or die "failed to flock file:$fn.tmp:$!";
        rename "$fn.tmp", $fn
            or die "failed to rename file:$fn.tmp to $fn:$!";
        $self->{fh} = $fh;
        $self->{id_for_fh} = $id;
    }
    # write to file with size of the status and its checksum
    seek $self->{fh}, 0, SEEK_SET
        or die "seek failed:$!";
    print {$self->{fh}} (
        md5($status),
        pack("N", length $status),
        $status,
    );
}

sub read_all {
    my $self = shift;
    my %ret;
    $self->_for_all(
        sub {
            my ($id, $fh) = @_;
            # detect collision using md5
            for (1..10) {
                seek $fh, 0, SEEK_SET
                    or die "seek failed:$!";
                my $data = do { local $/; join '', <$fh> };
                # silently ignore if data is too short
                return if length($data) < 16 + 4;
                # parse input
                my $md5 = substr($data, 0, 16);
                my $size = unpack("N", substr($data, 16, 4));
                my $status = substr($data, 20, $size);
                # compare md5 to detect collision
                next
                    if md5($status) ne $md5;
                # have read correct data, save and return
                $ret{$id} = $status;
                return;
            }
            # failed to read data in 10 consecutive attempts, bug?
            warn "failed to read status of id:$id, skipping";
        }
    );
    \%ret;
}

sub cleanup {
    my $self = shift;
    $self->_for_all(sub {});
}

sub _for_all {
    my ($self, $cb) = @_;
    my @files = glob "$self->{base_dir}/status_*";
    for my $fn (@files) {
        # obtain id from filename (or else ignore);
        # skip .tmp to avoid racing with update()
        $fn =~ m|/status_(.*)(?
            or next;
        my $id = $1;
        # ignore files removed after glob but before open
        open my $fh, '+<', $fn
            or next;
        # check if the file is still opened by the owner process using flock
        if ($id ne $self->worker_id->() && flock $fh, LOCK_EX | LOCK_NB) {
            # the owner has died, remove status file
            close $fh;
            unlink $fn
                or
                    not $!{ENOENT} and
                    warn "failed to remove an obsolete scoreboard file:$fn:$!";
            next;
        }
        # invoke
        $cb->($id, $fh);
        # close
        close $fh;
    }
}

sub _build_filename {
    my $self = shift;
    return "$self->{base_dir}/status_" . $self->worker_id->();
}

1;
__END__