File Coverage

blib/lib/Directory/Queue/Normal.pm

Criterion	Covered	Total	%
statement	276	329	83.8
branch	104	184	56.5
condition	6	24	25.0
subroutine	33	36	91.6
pod	9	9	100.0
total	428	582	73.5

#+##############################################################################
#                                                                              #
# File: Directory/Queue/Normal.pm                                              #
#                                                                              #
# Description: object oriented interface to a normal directory based queue     #
#                                                                              #
#-##############################################################################

#
# module definition
#

package Directory::Queue::Normal;
use strict;
use warnings;
our $VERSION  = "2.2";
our $REVISION = sprintf("%d.%02d", q$Revision: 1.16 $ =~ /(\d+)\.(\d+)/);

#
# used modules
#

use Directory::Queue qw(_create _name _touch /Regexp/ /special/);
use Encode qw(encode decode FB_CROAK LEAVE_SRC);
use No::Worries::Die qw(dief);
use No::Worries::File qw(file_read file_write);
use No::Worries::Stat qw(ST_MTIME ST_NLINK);
use No::Worries::Warn qw(warnf);
use POSIX qw(:errno_h);

#
# inheritance
#

our(@ISA) = qw(Directory::Queue);

#
# constants
#

# name of the directory holding temporary elements
use constant TEMPORARY_DIRECTORY => "temporary";

# name of the directory holding obsolete elements
use constant OBSOLETE_DIRECTORY => "obsolete";

# name of the directory indicating a locked element
use constant LOCKED_DIRECTORY => "locked";

#
# global variables
#

our(
    $_FileRegexp,     # regexp matching a file in an element directory
    %_Byte2Esc,       # byte to escape map
    %_Esc2Byte,       # escape to byte map
);

$_FileRegexp = qr/[0-9a-zA-Z]+/;
%_Byte2Esc   = ("\x5c" => "\\\\", "\x09" => "\\t", "\x0a" => "\\n");
%_Esc2Byte   = reverse(%_Byte2Esc);

#+++############################################################################
#                                                                              #
# Helper Functions                                                             #
#                                                                              #
#---############################################################################

#
# transform a hash of strings into a string (reference)
#
# note:
#  - the keys are sorted so that identical hashes yield to identical strings
#

sub _hash2string ($) {
    my($hash) = @_;
    my($value, $string);

    $string = "";
    foreach my $key (sort(keys(%{ $hash }))) {
        $value = $hash->{$key};
        dief("undefined hash value: %s", $key) unless defined($value);
        dief("invalid hash scalar: %s", $value) if ref($value);
        $key   =~ s/([\x5c\x09\x0a])/$_Byte2Esc{$1}/g;
        $value =~ s/([\x5c\x09\x0a])/$_Byte2Esc{$1}/g;
        $string .= $key . "\x09" . $value . "\x0a";
    }
    return(\$string);
}

#
# transform a string (reference) into a hash of strings
#
# note:
#  - duplicate keys are not checked (the last one wins)
#

sub _string2hash ($) {
    my($stringref) = @_;
    my($key, $value, %hash);

    foreach my $line (split(/\x0a/, ${ $stringref })) {
        if ($line =~ /^([^\x09\x0a]*)\x09([^\x09\x0a]*)$/o) {
            ($key, $value) = ($1, $2);
        } else {
            dief("unexpected hash line: %s", $line);
        }
        $key   =~ s/(\\[\\tn])/$_Esc2Byte{$1}/g;
        $value =~ s/(\\[\\tn])/$_Esc2Byte{$1}/g;
        $hash{$key} = $value;
    }
    return(\%hash);
}

#
# check if a path is old enough:
#  - return true if the path exists and is (strictly) older than the given time
#  - return false if it does not exist or it is newer
#  - die in case of any other error
#
# note:
#  - lstat() is used so symlinks are not followed
#

sub _older ($$) {
    my($path, $time) = @_;
    my(@stat);

    @stat = lstat($path);
    unless (@stat) {
        dief("cannot lstat(%s): %s", $path, $!) unless $! == ENOENT;
        # RACE: this path does not exist (anymore)
        return(0);
    }
    return($stat[ST_MTIME] < $time);
}

#
# count the number of sub-directories in the given directory:
#  - return undef if the directory does not exist (anymore)
#  - die in case of any other error
#

# stat version (faster):
#  - lstat() is used so symlinks are not followed
#  - this only checks the number of hard links
#  - we do not even check that the given path indeed points to a directory!
#  - this will return incorrect results on some filesystems like DOS or Btrfs

sub _subdirs_stat ($) {
    my($path) = @_;
    my(@stat);

    @stat = lstat($path);
    unless (@stat) {
        dief("cannot lstat(%s): %s", $path, $!) unless $! == ENOENT;
        # RACE: this path does not exist (anymore)
        return();
    }
    return($stat[ST_NLINK] - 2);
}

# readdir version (slower):
#  - we really count the number of entries
#  - we however do not check that these entries are themselves indeed directories
#  - this is the default method to favor correctness over speed

sub _subdirs_readdir ($) {
    my($path) = @_;

    return(scalar(_special_getdir($path)));
}

#
# wrapper method
#

sub _subdirs ($$) {
    my($self, $path) = @_;

    return($self->{nlink} ? _subdirs_stat($path) : _subdirs_readdir($path));
}

#
# check the given string to make sure it represents a valid element name
#

sub _check_element ($) {
    my($element) = @_;

    dief("invalid element: %s", $element)
        unless $element =~ m/^(?:$_DirectoryRegexp)\/(?:$_ElementRegexp)$/o;
}

#+++############################################################################
#                                                                              #
# Object Oriented Interface                                                    #
#                                                                              #
#---############################################################################

#
# object constructor
#

sub new : method {
    my($class, %option) = @_;
    my($self, $path, $options);

    # default object
    $self = __PACKAGE__->SUPER::_new(%option);
    foreach my $name (qw(path maxlock maxtemp rndhex umask)) {
        delete($option{$name});
    }
    # default options
    $self->{maxelts} = 16_000; # maximum number of elements per directory
    # check maxelts
    if (defined($option{maxelts})) {
        dief("invalid maxelts: %s", $option{maxelts})
            unless $option{maxelts} =~ /^\d+$/ and $option{maxelts} > 0;
        $self->{maxelts} = delete($option{maxelts});
    }
    # check nlink
    $self->{nlink} = delete($option{nlink});
    # check schema
    if (defined($option{schema})) {
        dief("invalid schema: %s", $option{schema})
            unless ref($option{schema}) eq "HASH";
        foreach my $name (keys(%{ $option{schema} })) {
            dief("invalid schema name: %s", $name)
                unless $name =~ /^($_FileRegexp)$/
                   and $name ne LOCKED_DIRECTORY;
            if ($option{schema}{$name} =~
                /^(binary|string|table)([\?\*]{0,2})$/) {
                $self->{type}{$name} = $1;
                $options = $2;
            } else {
                dief("invalid schema type: %s", $option{schema}{$name});
            }
            $self->{mandatory}{$name} = 1 unless $options =~ /\?/;
            $self->{ref}{$name} = 1 if $options =~ /\*/;
            dief("invalid schema type: %s", $option{schema}{$name})
                if $self->{type}{$name} eq "table" and $self->{ref}{$name};
        }
        dief("invalid schema: no mandatory data")
            unless $self->{mandatory};
        delete($option{schema});
    }
    # check unexpected options
    foreach my $name (keys(%option)) {
        dief("unexpected option: %s", $name);
    }
    # create directories
    foreach my $name (TEMPORARY_DIRECTORY, OBSOLETE_DIRECTORY) {
        $path = $self->{path}."/".$name;
        _special_mkdir($path, $self->{umask}) unless -d $path;
    }
    # so far so good...
    return($self);
}

#
# return the number of elements in the queue, regardless of their state
#

sub count : method {
    my($self) = @_;
    my($count, @list, $subdirs);

    $count = 0;
    # get the list of existing directories
    foreach my $name (_special_getdir($self->{path}, "strict")) {
        push(@list, $1) if $name =~ /^($_DirectoryRegexp)$/o; # untaint
    }
    # count sub-directories
    foreach my $name (@list) {
        $subdirs = _subdirs($self, $self->{path}."/".$name);
        $count += $subdirs if $subdirs;
    }
    # that's all
    return($count);
}

#
# check if an element is locked:
#  - this is best effort only as it may change while we test (only locking is atomic)
#  - if given a time, only return true on locks older than this time (needed by purge)
#

sub _is_locked ($$;$) {
    my($self, $name, $time) = @_;
    my($path, @stat);

    $path = $self->{path}."/".$name;
    return(0) unless -d $path."/".LOCKED_DIRECTORY;
    return(1) unless defined($time);
    @stat = lstat($path);
    unless (@stat) {
        dief("cannot lstat(%s): %s", $path, $!) unless $! == ENOENT;
        # RACE: this path does not exist (anymore)
        return(0);
    }
    return($stat[ST_MTIME] < $time);
}

#
# lock an element:
#  - return true on success
#  - return false in case the element could not be locked (in permissive mode)
#
# note:
#  - locking can fail:
#     - if the element has been locked by somebody else (EEXIST)
#     - if the element has been removed by somebody else (ENOENT)
#  - if the optional second argument is true, it is not an error if
#    the element cannot be locked (= permissive mode), this is the default
#    as one usually cannot be sure that nobody else will try to lock it
#  - the directory's mtime will change automatically (after a successful mkdir()),
#    this will later be used to detect stalled locks
#

sub lock : method { ## no critic 'ProhibitBuiltinHomonyms'
    my($self, $element, $permissive) = @_;
    my($path, $oldumask, $success);

    _check_element($element);
    $permissive = 1 unless defined($permissive);
    $path = $self->{path}."/".$element."/".LOCKED_DIRECTORY;
    if (defined($self->{umask})) {
        $oldumask = umask($self->{umask});
        $success = mkdir($path);
        umask($oldumask);
    } else {
        $success = mkdir($path);
    }
    unless ($success) {
        if ($permissive) {
            # RACE: the locked directory already exists
            return(0) if $! == EEXIST;
            # RACE: the element directory does not exist anymore
            return(0) if $! == ENOENT;
        }
        # otherwise this is unexpected...
        dief("cannot mkdir(%s): %s", $path, $!);
    }
    $path = $self->{path}."/".$element;
    unless (lstat($path)) {
        if ($permissive and $! == ENOENT) {
            # RACE: the element directory does not exist anymore
            # (this can happen if an other process locked & removed the element
            #  while our mkdir() was in progress... yes, this can happen!)
            return(0);
        }
        # otherwise this is unexpected...
        dief("cannot lstat(%s): %s", $path, $!);
    }
    # so far so good
    return(1);
}

#
# unlock an element:
#  - return true on success
#  - return false in case the element could not be unlocked (in permissive mode)
#
# note:
#  - unlocking can fail:
#     - if the element has been unlocked by somebody else (ENOENT)
#     - if the element has been removed by somebody else (ENOENT)
#  - if the optional second argument is true, it is not an error if
#    the element cannot be unlocked (= permissive mode), this is _not_ the default
#    as unlock() should normally be called by whoever locked the element
#

sub unlock : method {
    my($self, $element, $permissive) = @_;
    my($path);

    _check_element($element);
    $path = $self->{path}."/".$element."/".LOCKED_DIRECTORY;
    unless (rmdir($path)) {
        if ($permissive) {
            # RACE: the element directory or its lock does not exist anymore
            return(0) if $! == ENOENT;
        }
        # otherwise this is unexpected...
        dief("cannot rmdir(%s): %s", $path, $!);
    }
    # so far so good
    return(1);
}

#
# touch an element to indicate that it is still being used
#

sub touch : method {
    my($self, $element) = @_;

    _touch($self->{"path"}."/".$element);
}

#
# remove a locked element from the queue
#

sub remove : method {
    my($self, $element) = @_;
    my($temp, $path);

    _check_element($element);
    dief("cannot remove %s: not locked", $element)
        unless _is_locked($self, $element);
    # move the element out of its intermediate directory
    $path = $self->{path}."/".$element;
    while (1) {
        $temp = $self->{path}
           ."/".OBSOLETE_DIRECTORY
           ."/"._name($self->{rndhex});
        rename($path, $temp) and last;
        dief("cannot rename(%s, %s): %s", $path, $temp, $!)
            unless $! == ENOTEMPTY or $! == EEXIST;
        # RACE: the target directory was already present...
    }
    # remove the data files
    foreach my $name (_special_getdir($temp, "strict")) {
        next if $name eq LOCKED_DIRECTORY;
        if ($name =~ /^($_FileRegexp)$/o) {
            $path = $temp."/".$1; # untaint
        } else {
            dief("unexpected file in %s: %s", $temp, $name);
        }
        unlink($path) and next;
        dief("cannot unlink(%s): %s", $path, $!);
    }
    # remove the locked directory
    $path = $temp."/".LOCKED_DIRECTORY;
    while (1) {
        rmdir($path) or dief("cannot rmdir(%s): %s", $path, $!);
        rmdir($temp) and return;
        dief("cannot rmdir(%s): %s", $temp, $!)
            unless $! == ENOTEMPTY or $! == EEXIST;
        # RACE: this can happen if an other process managed to lock this element
        # while it was being removed (see the comment in the lock() method)
        # so we try to remove the lock again and again...
    }
}

#
# read a binary file and return a reference to the corresponding data
#

sub _file_read_bin ($) {
    my($path) = @_;
    my($data);

    file_read($path, data => \$data);
    return(\$data);
}

#
# read a UTF-8 encoded file and return a reference to the corresponding string
#

sub _file_read_utf8 ($) {
    my($path) = @_;
    my($data, $string);

    file_read($path, data => \$data);
    eval {
        local $SIG{__WARN__} = sub { die($_[0]) };
        $string = decode("UTF-8", $data, FB_CROAK);
    };
    return(\$string) unless $@;
    $@ =~ s/\s+at\s.+?\sline\s+\d+\.?$//;
    dief("cannot UTF-8 decode %s: %s", $path, $@);
}

#
# get an element from a locked element
#

sub get : method {
    my($self, $element) = @_;
    my(%data, $path, $ref);

    dief("unknown schema") unless $self->{type};
    _check_element($element);
    dief("cannot get %s: not locked", $element)
        unless _is_locked($self, $element);
    foreach my $name (keys(%{ $self->{type} })) {
        $path = "$self->{path}/$element/$name";
        unless (lstat($path)) {
            dief("cannot lstat(%s): %s", $path, $!) unless $! == ENOENT;
            if ($self->{mandatory}{$name}) {
                dief("missing data file: %s", $path);
            } else {
                next;
            }
        }
        if ($self->{type}{$name} =~ /^(binary|string)$/) {
            if ($self->{type}{$name} eq "string") {
                $ref = _file_read_utf8($path);
            } else {
                $ref = _file_read_bin($path);
            }
            $data{$name} = $self->{ref}{$name} ? $ref : ${ $ref };
        } elsif ($self->{type}{$name} eq "table") {
            $data{$name} = _string2hash(_file_read_utf8($path));
        } else {
            dief("unexpected data type: %s", $self->{type}{$name});
        }
    }
    return(\%data) unless wantarray();
    return(%data);
}

#
# return the name of the intermediate directory that can be used for insertion:
#  - if there is none, an initial one will be created
#  - if it is full, a new one will be created
#  - in any case the name will match $_DirectoryRegexp
#

sub _insertion_directory ($) {
    my($self) = @_;
    my(@list, $new, $subdirs);

    # get the list of existing directories
    foreach my $name (_special_getdir($self->{path}, "strict")) {
        push(@list, $1) if $name =~ /^($_DirectoryRegexp)$/o; # untaint
    }
    # handle the case with no directories yet
    unless (@list) {
        $new = sprintf("%08x", 0);
        _special_mkdir($self->{path}."/".$new, $self->{umask});
        return($new);
    }
    # check the last directory
    @list = sort(@list);
    $new = pop(@list);
    $subdirs = _subdirs($self, $self->{path}."/".$new);
    if (defined($subdirs)) {
        return($new) if $subdirs < $self->{maxelts};
        # this last directory is now full... create a new one
    } else {
        # RACE: at this point, the directory does not exist anymore, so it
        # must have been purged after we listed the directory contents...
        # we do not try to do more and simply create a new directory
    }
    # we need a new directory
    $new = sprintf("%08x", hex($new) + 1);
    _special_mkdir($self->{path}."/".$new, $self->{umask});
    return($new);
}

#
# add data to a directory
#

sub _add_data ($$$) {
    my($self, $data, $tempdir) = @_;
    my($ref, $utf8, $tmp, $path, $fh);

    foreach my $name (keys(%{ $data })) {
        dief("unexpected data: %s", $name) unless $self->{type}{$name};
        if ($self->{type}{$name} =~ /^(binary|string)$/) {
            if ($self->{ref}{$name}) {
                dief("unexpected %s data in %s: %s",
                     $self->{type}{$name}, $name, $data->{$name})
                    unless ref($data->{$name}) eq "SCALAR";
                $ref = $data->{$name};
            } else {
                dief("unexpected %s data in %s: %s",
                     $self->{type}{$name}, $name, $data->{$name})
                    if ref($data->{$name});
                $ref = \$data->{$name};
            }
            $utf8 = $self->{type}{$name} eq "string";
        } elsif ($self->{type}{$name} eq "table") {
            dief("unexpected %s data in %s: %s",
                 $self->{type}{$name}, $name, $data->{$name})
                unless ref($data->{$name}) eq "HASH";
            $ref = _hash2string($data->{$name});
            $utf8 = 1;
        } else {
            dief("unexpected data type in %s: %s",
                 $name, $self->{type}{$name});
        }
        if ($utf8) {
            eval {
                $tmp = encode("UTF-8", ${ $ref }, FB_CROAK|LEAVE_SRC);
            };
            if ($@) {
                $@ =~ s/\s+at\s.+?\sline\s+\d+\.?$//;
                dief("unexpected character in %s: %s", $name, $@);
            }
            $ref = \$tmp;
        }
        $path = "$tempdir/$name";
        $fh = _create($path, $self->{umask}, "strict");
        file_write($path, handle => $fh, data => $ref);
    }
}

#
# add a new element to the queue and return its name
#
# note:
#  - the destination directory must _not_ be created beforehand as it would
#    be seen as a valid (but empty) element directory by an other process,
#    we therefor use rename() from a temporary directory
#

sub add : method {
    my($self, @data) = @_;
    my($data, $tempdir, $dir, $new, $path, $ref, $utf8);

    dief("unknown schema") unless $self->{type};
    if (@data == 1) {
        $data = $data[0];
    } else {
        $data = { @data };
    }
    foreach my $name (keys(%{ $self->{mandatory} })) {
        dief("missing mandatory data: %s", $name)
            unless defined($data->{$name});
    }
    while (1) {
        $tempdir = $self->{path}
           ."/".TEMPORARY_DIRECTORY
           ."/"._name($self->{rndhex});
        last if _special_mkdir($tempdir, $self->{umask});
    }
    _add_data($self, $data, $tempdir);
    $dir = _insertion_directory($self);
    while (1) {
        $new = $dir."/"._name($self->{rndhex});
        $path = $self->{path}."/".$new;
        rename($tempdir, $path) and return($new);
        dief("cannot rename(%s, %s): %s", $tempdir, $path, $!)
            unless $! == ENOTEMPTY or $! == EEXIST;
        # RACE: the target directory was already present...
    }
}

#
# return the list of volatile (i.e. temporary or obsolete) directories
#

sub _volatile ($) {
    my($self) = @_;
    my(@list);

    foreach my $name (_special_getdir($self->{path} .
                                      "/" . TEMPORARY_DIRECTORY)) {
        push(@list, TEMPORARY_DIRECTORY."/".$1)
            if $name =~ /^($_ElementRegexp)$/o; # untaint
    }
    foreach my $name (_special_getdir($self->{path} .
                                      "/" . OBSOLETE_DIRECTORY)) {
        push(@list, OBSOLETE_DIRECTORY."/".$1)
            if $name =~ /^($_ElementRegexp)$/o; # untaint
    }
    return(@list);
}

#
# destroy a volatile directory
#

sub _destroy_dir ($) {
    my($dir) = @_;
    my($path);

    foreach my $name (_special_getdir($dir)) {
        next if $name eq LOCKED_DIRECTORY;
        $path = $dir."/".$name;
        unlink($path) and next;
        dief("cannot unlink(%s): %s", $path, $!) unless $! == ENOENT;
    }
    _special_rmdir($dir."/".LOCKED_DIRECTORY);
    _special_rmdir($dir);
}

#
# purge the queue:
#  - delete unused intermediate directories
#  - delete too old temporary directories
#  - unlock too old locked directories
#
# note: this uses first()/next() to iterate so this will reset the cursor
#

sub purge : method {
    my($self, %option) = @_;
    my(@list, $path, $subdirs, $oldtime, $locked);

    # check options
    $option{maxtemp} = $self->{maxtemp} unless defined($option{maxtemp});
    $option{maxlock} = $self->{maxlock} unless defined($option{maxlock});
    foreach my $name (keys(%option)) {
        dief("unexpected option: %s", $name)
            unless $name =~ /^(maxtemp|maxlock)$/;
        dief("invalid %s: %s", $name, $option{$name})
            unless $option{$name} =~ /^\d+$/;
    }
    # get the list of intermediate directories
    @list = ();
    foreach my $name (_special_getdir($self->{path}, "strict")) {
        push(@list, $1) if $name =~ /^($_DirectoryRegexp)$/o; # untaint
    }
    # try to purge all but the last intermediate directory
    if (@list > 1) {
        @list = sort(@list);
        pop(@list);
        foreach my $name (@list) {
            $path = $self->{path}."/".$name;
            $subdirs = _subdirs($self, $path);
            next if $subdirs or not defined($subdirs);
            _special_rmdir($path);
        }
    }
    # remove the volatile directories which are too old
    if ($option{maxtemp}) {
        $oldtime = time() - $option{maxtemp};
        foreach my $name (_volatile($self)) {
            $path = $self->{path}."/".$name;
            next unless _older($path, $oldtime);
            warnf("removing too old volatile element: %s", $name);
            _destroy_dir($path);
        }
    }
    # iterate to find abandoned locked entries
    if ($option{maxlock}) {
        $oldtime = time() - $option{maxlock};
        $locked = $self->first();
        while ($locked) {
            next unless _is_locked($self, $locked, $oldtime);
            warnf("removing too old locked element: %s", $locked);
            $self->unlock($locked, 1);
        } continue {
            $locked = $self->next();
        }
    }
}

1;

__END__