File Coverage

blib/lib/Sub/Throttler/Periodic/EV.pm

Criterion	Covered	Total	%
statement	92	102	90.2
branch	27	36	75.0
condition	5	10	50.0
subroutine	20	21	95.2
pod	8	8	100.0
total	152	177	85.8

package Sub::Throttler::Periodic::EV;
use 5.010001;
use warnings;
use strict;
use utf8;
use Carp;
our @CARP_NOT = qw( Sub::Throttler );

our $VERSION = 'v0.2.10';

use parent qw( Sub::Throttler::Limit );
use Sub::Throttler qw( throttle_flush );
use Time::HiRes qw( time sleep );
use Scalar::Util qw( weaken );
use EV;


sub new {
    use warnings FATAL => qw( misc );
    my ($class, %opt) = @_;
    my $self = bless {
        limit   => delete $opt{limit} // 1,
        period  => delete $opt{period} // 1,
        acquired=> {},  # { $id => { $key => $quantity, … }, … }
        used    => {},  # { $key => $quantity, … }
        }, ref $class || $class;
    croak 'limit must be an unsigned integer' if $self->{limit} !~ /\A\d+\z/ms;
    croak 'period must be a positive number' if $self->{period} <= 0;
    croak 'bad param: '.(keys %opt)[0] if keys %opt;
    weaken(my $this = $self);
    $self->{_t} = EV::periodic 0, $self->{period}, 0, sub { $this && $this->_tick() };
    $self->{_t}->keepalive(0);
    return $self;
}

sub acquire {
    my ($self, $id, $key, $quantity) = @_;
    if (!$self->try_acquire($id, $key, $quantity)) {
        if ($quantity <= $self->{limit} && $self->{used}{$key}) {
            my $time = time;
            my $delay = int($time/$self->{period})*$self->{period} + $self->{period} - $time;
            sleep $delay;
            $self->_tick();
        }
        if (!$self->try_acquire($id, $key, $quantity)) {
            croak "$self: unable to acquire $quantity of resource '$key'";
        }
    }
    return $self;
}

sub load {
    my ($class, $state) = @_;
    croak 'bad state: wrong algorithm' if $state->{algo} ne __PACKAGE__;
    my $v = version->parse($state->{version});
    if ($v > $VERSION) {
        carp 'restoring state saved by future version';
    }
    my $self = $class->new(limit=>$state->{limit}, period=>$state->{period});
    # time jump backward, no matter how much, handled like we still is in
    # current period, to be safe
    if (int($state->{at}/$self->{period})*$self->{period} + $self->{period} > time) {
        $self->{used} = $state->{used};
    }
    if (keys %{ $self->{used} }) {
        $self->{_t}->keepalive(1);
    }
    return $self;
}

sub period {
    my ($self, $period) = @_;
    if (1 == @_) {
        return $self->{period};
    }
    croak 'period must be a positive number' if $period <= 0;
    $self->{period} = $period;
    $self->{_t}->set(0, $self->{period}, 0);
    return $self;
}

sub release {
    my ($self, $id) = @_;
    croak sprintf '%s not acquired anything', $id if !$self->{acquired}{$id};
    delete $self->{acquired}{$id};
    return $self;
}

sub release_unused {
    my $self = shift->SUPER::release_unused(@_);
    if (!keys %{ $self->{used} }) {
        $self->{_t}->keepalive(0);
    }
    return $self;
}

sub save {
    my ($self) = @_;
    my $state = {
        algo    => __PACKAGE__,
        version => version->declare($VERSION)->numify,
        limit   => $self->{limit},
        period  => $self->{period},
        used    => $self->{used},
        at      => time,
    };
    return $state;
}

sub try_acquire {
    my $self = shift;
    if ($self->SUPER::try_acquire(@_)) {
        $self->{_t}->keepalive(1);
        return 1;
    }
    return;
}

sub _tick {
    my $self = shift;
    for my $id (keys %{ $self->{acquired} }) {
        for my $key (keys %{ $self->{acquired}{$id} }) {
            $self->{acquired}{$id}{$key} = 0;
        }
    }
    # OPTIMIZATION call throttle_flush() only if amount of available
    # resources increased (i.e. if some sources was released)
    if (keys %{ $self->{used} }) {
        $self->{used} = {};
        throttle_flush();
    }
    if (!keys %{ $self->{used} }) {
        $self->{_t}->keepalive(0);
    }
    return;
}


1; # Magic true value required at end of module
__END__