File Coverage

lib/Net/ZooIt.pm

Criterion	Covered	Total	%
statement	22	24	91.6
branch			n/a
condition			n/a
subroutine	8	8	100.0
pod			n/a
total	30	32	93.7

package Net::ZooIt;

use strict;
use warnings;

our $VERSION = '0.21';

use Sys::Hostname qw(hostname);
use Carp qw(croak);
use POSIX qw(strftime);
use Time::HiRes qw(time);
use feature ':5.10';

use Net::ZooKeeper qw(:all);

use base qw(Exporter);
our @EXPORT = qw(ZOOIT_DIE ZOOIT_ERR ZOOIT_WARN ZOOIT_INFO ZOOIT_DEBUG);

# Logging
sub ZOOIT_DIE { 0 }
sub ZOOIT_ERR { 1 }
sub ZOOIT_WARN { 2 }
sub ZOOIT_INFO { 3 }
sub ZOOIT_DEBUG { 4 }
my @log_levels = qw(ZOOIT_DIE ZOOIT_ERR ZOOIT_WARN ZOOIT_INFO ZOOIT_DEBUG);
my $log_level = 1;

sub set_log_level {
    my $level = shift;
    return unless $level =~ /^\d+$/;
    $log_level = $level;
}

sub logger {
    my ($level, $msg) = @_;
    return unless $level =~ /^\d+$/;
    return if $level > $log_level;
    $msg =~ s/\n$//;
    my $prefix = strftime '%Y-%m-%dT%H:%M:%SZ', gmtime;
    $prefix .= " $$";
    $prefix .= " $log_levels[$level]";
    print STDERR "$prefix $msg\n";
    croak $msg if $level == 0;
}

sub zdie { logger ZOOIT_DIE, @_ }
sub zerr { logger ZOOIT_ERR, @_ }
sub zwarn { logger ZOOIT_WARN, @_ }
sub zinfo { logger ZOOIT_INFO, @_ }
sub zdebug { logger ZOOIT_DEBUG, @_ }

sub z2txt {
    my ($type, $val) = @_;
    our %tv2txt;
    unless (%tv2txt) {
        foreach my $t (qw(errors events node_flags states)) {
            foreach my $name (@{$Net::ZooKeeper::EXPORT_TAGS{$t}}) {
                no strict "refs";
                my $code = &$name();
                use strict "refs";
                defined $code and $tv2txt{$t}->{$code} = $name;
            }
        }
    }
    my $t = {qw(
        err errors error errors
        ev events event events
        st states state states
        node node_flags flag node_flags
    )}->{$type} // $type;
    $tv2txt{$t} or zerr "z2txt: unknown type $type" and return;
    my $txt = $tv2txt{$t}->{$val} or return;
    return $txt;
}

sub zerr2txt {
    return z2txt 'errors', shift;
}

# Generate and split sequential znode names
sub gen_seq_name { hostname . ".PID.$$-" }
sub split_seq_name { shift =~ /^(.+-)(\d+)$/; $1, $2 }

# ZooKeeper recipe Lock
sub new_lock {
    my $class = shift;
    zerr "lock will be released immediately, new_lock called in void context"
        unless defined wantarray;
    my %p = @_;
    zdie "Param zk must be a connect Net::ZooKeeper object"
        unless ref $p{zk};
    zdie "Param path must be a valid ZooKeeper znode path"
        unless $p{path} =~ m|^/.+|;

    my $lockname = gen_seq_name;
    my $lock = $p{zk}->create(
        "$p{path}/$lockname" => 1,
        flags => ZOO_EPHEMERAL|ZOO_SEQUENCE,
        acl => ZOO_OPEN_ACL_UNSAFE,
    );
    unless ($lock) {
        zerr "Could not create $p{path}/$lockname: " . zerr2txt($p{zk}->get_error);
        return;
    }
    zinfo "Created lock $lock";
    # Create the blessed object now, for auto-deletion if next operations fail
    my $res = bless { lock => $lock, zk => $p{zk} }, $class;
    my $t0 = time;

    my ($basename, $n) = split_seq_name $res->{lock};
    while (1) {
        _gc($p{zk});

        my @locks = $p{zk}->get_children($p{path});
        my $err = $p{zk}->get_error;
        if ($err ne ZOK) {
            zerr "Could not get lock list: " . zerr2txt($err);
            return;
        }
        zdebug "Get lock list: @locks";
        # Look for other lock with highest sequence number lower than mine
        my ($lock_prev, $n_prev);
        foreach (@locks) {
            my ($basename_i, $n_i) = split_seq_name $_;
            next if $n_i >= $n;
            if (!defined $n_prev || $n_i > $n_prev) {
                $n_prev = $n_i;
                $lock_prev = $_;
            }
        }
        # If none found, the lock is mine
        unless (defined $n_prev) {
            zinfo "Take lock: $res->{lock}";
            return $res;
        }
        # I can't take lock, abort if timeout reached
        my $dt;
        if (defined $p{timeout}) {
            $dt = $t0 + $p{timeout} - time;
            if ($dt <= 0) {
                zinfo "Timeout reached, abort";
                return;
            }
        }
        # Wait for lock with highest seq number lower than mine to be deleted
        $dt //= 60;
        $dt *= 1000;
        my $w = $p{zk}->watch(timeout => $dt);
        if ($p{zk}->exists("$p{path}/$lock_prev", watch => $w)) {
            zinfo "Wait for delete $p{path}/$lock_prev";
            $w->wait;
            my $event = z2txt('ev', $w->{event}) // 'timeout';
            zinfo "Wait for delete $p{path}/$lock_prev over: $event";
        }
    }
}

# ZooKeeper recipe Queue
sub new_queue {
    my $class = shift;
    my %p = @_;
    zdie "Param zk must be a connect Net::ZooKeeper object"
        unless ref $p{zk};
    zdie "Param path must be a valid ZooKeeper znode path"
        unless $p{path} =~ m|^/.+|;

    return bless { queue => $p{path}, zk => $p{zk} }, $class;
}

sub put_queue {
    my ($self, $data) = @_;
    my $itemname = gen_seq_name;
    my $item = $self->{zk}->create(
        "$self->{queue}/$itemname" => $data,
        flags => ZOO_SEQUENCE,
        acl => ZOO_OPEN_ACL_UNSAFE,
    );
    unless ($item) {
        zerr "Could not create $self->{queue}/$itemname: " . zerr2txt($self->{zk}->get_error);
        return;
    }
    zinfo "Created queue item $item";
    return 1;
}

sub get_queue {
    my $self = shift;
    my %p = @_;
    my $t0 = time;
    while (1) {
        _gc($self->{zk});

        my @items = $self->{zk}->get_children($self->{queue});
        my $err = $self->{zk}->get_error;
        if ($err ne ZOK) {
            zerr "Could not get queue items: " . zerr2txt($err);
            return;
        }
        zdebug "Get queue items: @items";
        # Look for queue item with lowest seq number
        my ($item_min, $n_min);
        foreach (@items) {
            my ($item_i, $n_i) = split_seq_name $_;
            if (!defined $n_min || $n_i < $n_min) {
                $n_min = $n_i;
                $item_min = $_;
            }
        }
        # If queue empty, wait for get_children, max timeout [s]
        unless (defined $n_min) {
            my $dt;
            if (defined $p{timeout}) {
                $dt = $t0 + $p{timeout} - time;
                if ($dt <= 0) {
                    zinfo "Timeout reached, abort";
                    return;
                }
            }
            $dt //= 60;
            $dt *= 1000;
            my $w = $self->{zk}->watch(timeout => $dt);
            unless ($self->{zk}->get_children("$self->{queue}", watch => $w)) {
                zinfo "Wait for children in $self->{queue}";
                $w->wait;
                my $event = z2txt('ev', $w->{event}) // 'timeout';
                zinfo "Wait for children in $self->{queue} over: $event";
            }
            next;
        }
        # Get data, attempt to delete znode with lowest seq number
        zinfo "Attempt to get/delete $item_min";
        my $data = $self->{zk}->get("$self->{queue}/$item_min");
        $err = $self->{zk}->get_error;
        if ($err ne ZOK) {
            zerr "Could not get item data: " . zerr2txt($err);
            next;
        }
        if ($self->{zk}->delete("$self->{queue}/$item_min")) {
            return $data;
        }
        $err = $self->{zk}->get_error;
        if ($err ne ZNONODE) {
            zerr "Error deleting queue item: " . zerr2txt($err);
            return;
        } else {
            zinfo "Someone else deleted $item_min";
        }
    }
}

# Automatic deletion of znodes when ZooIt objects go out of scope
# Garbage collection for znodes deleted during ZCONNECTIONLOSS
my @garbage;

sub DESTROY {
    my $self = shift;
    if ($self->{lock}) {
        zinfo "DESTROY deleting lock: $self->{lock}";
        $self->{zk}->delete($self->{lock});
        my $err = zerr2txt($self->{zk}->get_error);
        if ($err ne 'ZOK') {
            push @garbage, $self->{lock};
            zerr "Could not delete $self->{lock}: $err";
        }
        delete $self->{lock};
    }
}

sub _gc {
    my $zk = shift;
    while (my $znode = shift @garbage) {
        zinfo "_gc deleting $znode";
        $zk->delete($znode);
        my $err = zerr2txt($zk->get_error);
        zdebug "  $err";
        if ($err eq 'ZOK' || $err eq 'ZNONODE') {
            zinfo "$znode deleted by _gc";
        } else {
            zerr "$znode could not be deleted by _gc: $err";
            unshift @garbage, $znode;
            last;
        }
    }
}

1;

__END__