File Coverage

blib/lib/Queue/Q/ReliableFIFO/Redis.pm

Criterion	Covered	Total	%
statement	33	422	7.8
branch	0	168	0.0
condition	0	112	0.0
subroutine	11	47	23.4
pod	23	24	95.8
total	67	773	8.6

package Queue::Q::ReliableFIFO::Redis;
use strict;
use warnings;
use Carp qw(croak cluck);

use parent 'Queue::Q::ReliableFIFO';
use Queue::Q::ReliableFIFO::Item;
use Queue::Q::ReliableFIFO::Lua;
use Redis;
use Time::HiRes qw(usleep);

use Class::XSAccessor {
    getters => [qw(
                server
                port
                db
                queue_name
                busy_expiry_time
                claim_wait_timeout
                requeue_limit
                redis_conn
                redis_options
                warn_on_requeue
                _main_queue
                _busy_queue
                _failed_queue
                _time_queue
                _temp_queue
                _log_queue
                _script_cache
                _lua
                )],
    setters => {
        set_requeue_limit => 'requeue_limit',
        set_busy_expiry_time => 'busy_expiry_time',
        set_claim_wait_timeout => 'claim_wait_timeout',
    }
};

my %QueueType = map { $_ => undef } (qw(main busy failed time temp log));

sub new {
    my ($class, %params) = @_;
    for (qw(server port queue_name)) {
        croak("Need '$_' parameter")
            if not defined $params{$_};
    }

    my %AllowedNewParams = map { $_ => undef } (qw(
        server port db queue_name busy_expiry_time
        claim_wait_timeout requeue_limit redis_conn redis_options
        warn_on_requeue));
    for (keys %params) {
        croak("Invalid parameter '$_'")
            if not exists $AllowedNewParams{$_};
    }

    my $self = bless({
        requeue_limit      => 5,
        busy_expiry_time   => 30,
        claim_wait_timeout => 1,
        db                 => 0,
        warn_on_requeue    => 0,
        %params
    } => $class);
    $self->{"_$_" . '_queue'} = $params{queue_name} . "_$_"
        for keys %QueueType;

    $self->{redis_options} ||= { reconnect => 60 };
    $self->{redis_conn} ||= Redis->new(
        # by default, auto-reconnect during 60 seconds
        %{$self->{redis_options}},
        encoding => undef, # force undef for binary data
        server => join(":", $self->server, $self->port),
    );

    $self->redis_conn->select($self->db) if $self->db;

    $self->{_lua}
        = Queue::Q::ReliableFIFO::Lua->new(redis_conn => $self->redis_conn);

    return $self;
}

sub clone {
    my ($class, $org, %params) = @_;
    my %default = map { $_ => $org->{$_} }
                  grep m/^[a-zA-Z]/,
                  keys %$org;
    return $class->new(%default, %params);
}

sub enqueue_item {
    my $self = shift;
    return if not @_;

    return $self->redis_conn->lpush(
        $self->_main_queue,
        map { Queue::Q::ReliableFIFO::Item->new(data => $_)->_serialized } @_
    );
}

use constant NONBLOCKING => 0;
use constant BLOCKING => 1;

sub claim_item {
    my ($self, $n) = @_;
    return $self->_claim_item_internal($n, BLOCKING);
}

sub claim_item_nonblocking {
    my ($self, $n) = @_;
    return $self->_claim_item_internal($n, NONBLOCKING);
}

sub _claim_item_internal {
    my ($self, $n, $doblocking) = @_;
    $n ||= 1;
    my $timeout = $self->claim_wait_timeout;
    if ($n == 1) {
        # rpoplpush gives higher throughput than the blocking version
        # (i.e. brpoplpush). So use the blocked version only when we
        # need to wait.
        my $value;
        $value = $self->redis_conn->rpoplpush($self->_main_queue, $self->_busy_queue);
        if (not defined($value) and $doblocking == BLOCKING) {
            $value = $self->redis_conn->brpoplpush($self->_main_queue, $self->_busy_queue, $timeout);
        }
        return if not $value;
        my $item;
        eval { ($item) = Queue::Q::ReliableFIFO::Item->new(_serialized => $value); };
        # FIXME ignoring exception in eval{}!
        return $item;
    }
    else {
        my $conn = $self->redis_conn;
        my $qn = $self->_main_queue;
        my $bq = $self->_busy_queue;
        my @items;
        my $serial;
        if ($n > 30) {
            # yes, there is a race, but it's an optimization only
            my ($l) = $self->redis_conn->llen($qn);
            $n = $l if $l < $n;
        }
        eval {
            $conn->rpoplpush($qn, $bq, sub {
                if (defined $_[0]) {
                    push @items,
                    Queue::Q::ReliableFIFO::Item->new(_serialized => $_[0])
                }
            }) for 1..$n;
            $conn->wait_all_responses;
            if (@items == 0 && $doblocking == BLOCKING) {
                # list seems empty, use the blocking version
                $serial = $conn->brpoplpush($qn, $bq, $timeout);
                if (defined $serial) {
                    push(@items,
                        Queue::Q::ReliableFIFO::Item->new(_serialized => $serial));
                    undef $serial;
                    $conn->rpoplpush($qn, $bq, sub {
                        if (defined $_[0]) {
                            push @items,
                                Queue::Q::ReliableFIFO::Item->new(
                                    _serialized => $_[0]);
                        }
                    }) for 1 .. ($n-1);
                    $conn->wait_all_responses;
                }
            }
            1;
        }
        or do {
            return @items;  # return with whatever we have...
        };
        return @items;
    }
}

sub mark_item_as_done {
    my $self = shift;
    if (@_ == 1) {
        return $self->redis_conn->lrem(
            $self->_busy_queue, -1, $_[0]->_serialized);
    }
    else {
        # TODO since lrem is an O(n) operation in size of busy list,
        #      there's a crossover point at which having l items to remove
        #      from said list is better done in a single O(n) loop through
        #      the list (in Lua?) rather than in l*O(n)=O(ln) operations via
        #      _lrem!
        my $conn = $self->redis_conn;
        my $count = 0;
        $conn->lrem(
            $self->_busy_queue, -1, $_->_serialized, sub { $count += $_[0] })
                for @_;
        $conn->wait_all_responses;
        return $count;
    }
}

sub unclaim  {
    my $self = shift;
    return $self->__requeue_busy(1, undef, @_);
}

sub requeue_busy_item {
    my ($self, $raw) = @_;
    return $self->__requeue_busy(0, undef, $raw);
}

sub requeue_busy {
    my $self = shift;
    return $self->__requeue_busy(0, undef, @_);
}

sub requeue_busy_error {
    my $self = shift;
    my $error= shift;
    return $self->__requeue_busy(0, $error, @_);
}

sub __requeue_busy  {
    my $self = shift;
    my $place = shift;  # 0: producer side, 1: consumer side
    my $error = shift;  # error message
    my $n = 0;
    eval {
        $n += $self->_lua->call(
            'requeue_busy',
            3,
            $self->_busy_queue,
            $self->_main_queue,
            $self->_failed_queue,
            time(),
            $_->_serialized,
            $self->requeue_limit,
            $place,
            $error || '',
        ) for @_;
        1;
    }
    or do {
        cluck("Lua call went wrong! $@");
    };
    return $n;
}

sub requeue_failed_item {
    #
    # **deprecated ***
    # This can stress Redis very hard when there are many failed items.
    # The lrem operation does a scan. If the item is not
    # at the position where the lrem-search start, the scan goes on.
    # A sleep is added in case the method is called for multiple items.
    #
    my $self = shift;
    my $n = 0;
    eval {
        for (@_) {
            $n += $self->_lua->call(
                'requeue_failed_item',
                2,
                $self->_failed_queue,
                $self->_main_queue,
                time(),
                $_->_serialized,
            );
            usleep(1e5);
        }
        1;
    }
    or do {
        cluck("Lua call went wrong! $@");
    };
    return $n;
}

sub requeue_failed_items {
    my $self = shift;
    if (@_ == 1) {
        # old API
        my $limit = shift;
        my $n = $self->_lua->call(
            'requeue_failed',
            2,
            $self->_failed_queue,
            $self->_main_queue,
            time(),
            $limit
        );
        if (!defined $n) {
            cluck("Lua call went wrong! $@");
        }
        return $n;
    }
    my %options = @_;
    # delay: how long before trying again after a (temporary) fail
    my $delay   = delete $options{Delay}        || 0;
    my $max_fc  = delete $options{MaxFailCount} || -1;
    my $chunk   = delete $options{Chunk}        || 100;
    cluck("Invalid option: $_") for (keys %options);

    my $total_requeued = 0;
    if ($self->queue_length('failed') > 0) {
        my ($todo, $requeued) = (0,0);
        do {
            ($todo, $requeued) = split(/\s+/, $self->_lua->call(
                'requeue_failed_gentle',
                3,
                $self->_failed_queue,
                $self->_main_queue,
                $self->_temp_queue,
                time(),
                $chunk,
                $delay,
                $max_fc,
            ));
            $total_requeued += $requeued;
            usleep(1e5);
        }
        while($todo > 0);
    }
    return $total_requeued;
}

sub get_and_flush_failed_items {
    # depreacted, use remove_failed_items
    my ($self, %options) = @_;
    my (undef, $failures) = $self->remove_failed_items(%options);
    return @$failures;
}

sub remove_failed_items {
    my ($self, %options) = @_;
    my $min_age = delete $options{MinAge}       || 0;
    my $min_fc  = delete $options{MinFailCount} || 0;
    my $chunk   = delete $options{Chunk}        || 100;
    my $loglimit= delete $options{LogLimit}     || 100;
    cluck("Invalid option: $_") for (keys %options);

    my $total_removed= 0;
    if ($self->queue_length('failed') > 0) {
        my ($todo, $removed) = (0,0);
        do {
            my $now = time();
            ($todo, $removed) = split(/\s+/, $self->_lua->call(
                'remove_failed_gentle',
                3,
                $self->_failed_queue,
                $self->_temp_queue,
                $self->_log_queue,
                $now,
                $chunk,
                ($now - $min_age),
                $min_fc,
                $loglimit,
            ));
            $total_removed += $removed;
            usleep(1e5);
        }
        while($todo > 0);
    }
    return (0,[])
        if $total_removed == 0;

    my $conn = $self->redis_conn;
    my @serial =
        map { Queue::Q::ReliableFIFO::Item->new(_serialized => $_) }
        $conn->lrange($self->_log_queue, 0, -1);
    $conn->del($self->_log_queue);
    return ($total_removed, \@serial);
}

sub flush_queue {
    my $self = shift;
    my $conn = $self->redis_conn;
    $conn->multi;
    $conn->del($_)
        for ($self->_main_queue, $self->_busy_queue,
             $self->_failed_queue, $self->_time_queue);
    $conn->exec;
    return;
}

sub queue_length {
    my ($self, $type) = @_;
    __validate_type(\$type);
    my $qn = $self->queue_name . "_$type";
    my ($len) = $self->redis_conn->llen($qn);
    return $len;
}

sub peek_item {
    my ($self, $type) = @_;
    # this function returns the value of oldest item in the queue
    __validate_type(\$type);
    my $qn = $self->queue_name . "_$type";

    # take oldest item
    my ($serial) = $self->redis_conn->lrange($qn,-1,-1);
    return undef if ! $serial;    # empty queue

    my $item = Queue::Q::ReliableFIFO::Item->new(_serialized => $serial);
    return $item->data();
}

sub age {
    my ($self, $type) = @_;
    # this function returns age of oldest item in the queue (in seconds)
    __validate_type(\$type);
    my $qn = $self->queue_name . "_$type";

    # take oldest item
    my ($serial) = $self->redis_conn->lrange($qn,-1,-1);
    return 0 if ! $serial;    # empty queue, so age 0

    my $item = Queue::Q::ReliableFIFO::Item->new(_serialized => $serial);
    return time() - $item->time_queued;
}

sub raw_items_main {
    my $self = shift;
    return $self->_raw_items('main', @_);
}

sub raw_items_busy {
    my $self = shift;
    return $self->_raw_items('busy', @_);
}

sub raw_items_failed {
    my $self = shift;
    return $self->_raw_items('failed', @_);
}

sub _raw_items {
    my ($self, $type, $n) = @_;
    #__validate_type(\$type); # truism, cf. the ten lines above this
    $n ||= 0;
    my $qn = $self->queue_name . "_$type";
    return
        map { Queue::Q::ReliableFIFO::Item->new(_serialized => $_); }
        $self->redis_conn->lrange($qn, -$n, -1);
}

sub __validate_type {
    my $type = shift;
    $$type ||= 'main';
    croak("Unknown queue type $$type")
        if not exists $QueueType{$$type};
}

sub memory_usage_perc {
    my $self = shift;
    my $conn = $self->redis_conn;
    my $info = $conn->info('memory');
    my $mem_used = $info->{used_memory};
    my (undef, $mem_avail) = $conn->config('get', 'maxmemory');
    return 100 if $mem_avail == 0; # if nothing is available, it's full!
    return $mem_used * 100 / $mem_avail;
}


SCOPE: {
    my %ValidErrorActions = map { $_ => 1 } (qw(drop requeue));
    my %ValidOptions       = map { $_ => 1 } (qw(
        Chunk DieOnError ReturnOnDie MaxItems MaxSeconds ProcessAll Pause ReturnWhenEmpty NoSigHandlers WarnOnError
    ));

    sub consume {
        my ($self, $callback, $error_action, $options) = @_;
        # validation of input
        $error_action ||= 'requeue';
        croak("Unknown error action")
            if not exists $ValidErrorActions{$error_action};
        my %error_subs = (
            'drop'    => sub { my ($self, $item) = @_;
                                $self->mark_item_as_done($item); },
            'requeue' => sub { my ($self, $item, $error) = @_;
                               $self->requeue_busy_error($error, $item); },
        );
        my $onerror = $error_subs{$error_action}
            || croak("no handler for $error_action");

        $options = $options ? {%$options} : {};
        my $chunk       = delete $options->{Chunk} || 1;
        croak("Chunk should be a number > 0") if (! $chunk > 0);
        cluck("DieOnError is deprecated, use ReturnOnDie instead")
            if exists $options->{DieOnError};
        my $return      = delete $options->{ReturnOnDie} || delete $options->{DieOnError} || 0;
        my $maxitems    = delete $options->{MaxItems} || -1;
        my $maxseconds  = delete $options->{MaxSeconds} || 0;
        my $pause       = delete $options->{Pause} || 0;
        my $process_all = delete $options->{ProcessAll} || 0;
        my $return_when_empty= delete $options->{ReturnWhenEmpty} || 0;
        my $nohandlers  = delete $options->{NoSigHandlers} || 0;
        my $warn_on_error = delete $options->{WarnOnError} || 0;
        croak("Option ProcessAll without Chunk does not make sense")
            if $process_all && $chunk <= 1;
        croak("Option Pause without Chunk does not make sense")
            if $pause && $chunk <= 1;

        for (keys %$options) {
            croak("Unknown option $_") if not exists $ValidOptions{$_};
        }
        my $stop_time = $maxseconds > 0 ? time() + $maxseconds : 0;

        # Now we can start...
        my $stop = 0;
        my $MAX_RECONNECT = 60;
        my $sigint  = ref $SIG{INT}  eq 'CODE' ? $SIG{INT}  : undef;
        my $sigterm = ref $SIG{TERM} eq 'CODE' ? $SIG{TERM} : undef;
        local $SIG{INT} = $nohandlers ? $sigint : sub {
            print "stopping\n";
            $stop = 1;
            &$sigint if $sigint;
        };
        local $SIG{TERM} = $nohandlers ? $sigterm : sub {
            print "stopping\n";
            $stop = 1;
            &$sigterm if $sigterm;
        };

        if ($chunk == 1) {
            my $die_afterwards = 0;
            my $claimed_count = 0;
            my $done_count = 0;
            while(!$stop) {
                my $item = eval { $self->claim_item(); };
                if (!$item) {
                    last if $return_when_empty
                                || ($stop_time > 0 && time() >= $stop_time);
                    next;    # nothing claimed this time, try again
                }
                $claimed_count++;
                my $ok = eval { $callback->($item->data); 1; };
                if (!$ok) {
                    my $error = _clean_error($@);
                    warn "callback had an error: $error"
                      if $warn_on_error and $error;
                    for (1 .. $MAX_RECONNECT) {    # retry if connection is lost
                        eval { $onerror->($self, $item, $error); 1; }
                        or do {
                            last if $stop;
                            sleep 1;
                            next;
                        };
                        last;
                    }
                    if ($return) {
                        $stop = 1;
                        cluck("Stopping because of ReturnOnDie\n");
                    }
                } else {
                    for (1 .. $MAX_RECONNECT) {    # retry if connection is lost
                        eval {
                            $done_count += $self->mark_item_as_done($item);
                            1;
                        } or do {
                            last if $stop;
                            sleep 1;
                            next;
                        };
                        last;
                    }
                }
                $stop = 1 if ($maxitems > 0 && --$maxitems == 0)
                                || ($stop_time > 0 && time() >= $stop_time);
            }
            my $still_busy = $claimed_count - $done_count;
            warn "not all items removed from busy queue ($still_busy)\n"
                if $self->warn_on_requeue && $still_busy;
        }
        else {
            my $die_afterwards = 0;
            my $t0 = Time::HiRes::time();
            while(!$stop) {
                my @items;

                # give queue some time to grow
                if ($pause) {
                    my $pt = ($pause - (Time::HiRes::time()-$t0))*1e6;
                    Time::HiRes::usleep($pt) if $pt > 0;
                }

                eval { @items = $self->claim_item($chunk); 1; }
                or do {
                    print "error with claim\n";
                };
                $t0 = Time::HiRes::time() if $pause; # only relevant for pause
                if (@items == 0) {
                    last if $return_when_empty
                                || ($stop_time > 0 && time() >= $stop_time);
                    next;    # nothing claimed this time, try again
                }
                my @done;
                if ($process_all) {
                    # process all items in one call (option ProcessAll)
                    my $ok = eval { $callback->(map { $_->data } @items); 1; };
                    if ($ok) {
                        @done = splice @items;
                    }
                    else {
                        # we need to call onerror for all items now
                        @done = (); # consider all items failed
                        my $error = _clean_error($@);
                        warn "callback had an error: $error"
                          if $warn_on_error and $error;
                        while (my $item = shift @items) {
                            for (1 .. $MAX_RECONNECT) {
                                eval { $onerror->($self, $item, $error); 1; }
                                or do {
                                    last if $stop; # items might stay in busy mode
                                    sleep 1;
                                    next;
                                };
                                last;
                            }
                            if ($return) {
                                cluck("Stopping because of ReturnOnDie\n");
                                $stop = 1;
                            }
                            last if $stop;
                        }
                    }
                }
                else {
                    # normal case: process items one by one
                    while (my $item = shift @items) {
                        my $ok = eval { $callback->($item->data); 1; };
                        if ($ok) {
                            push @done, $item;
                        }
                        else {
                            my $error = _clean_error($@);
                            warn "callback had an error: $error"
                              if $warn_on_error and $error;
                            # retry if connection is lost
                            for (1 .. $MAX_RECONNECT) {
                                eval { $onerror->($self, $item, $error); 1; }
                                or do {
                                    last if $stop;
                                    sleep 1;
                                    next;
                                };
                                last;
                            }
                            if ($return) {
                                cluck("Stopping because of ReturnOnDie\n");
                                $stop = 1;
                            }
                        }
                        last if $stop;
                    }
                }
                my $count = 0;
                for (1 .. $MAX_RECONNECT) {
                    eval { $count += $self->mark_item_as_done(@done); 1; }
                    or do {
                        last if $stop;
                        sleep 1;
                        next;
                    };
                    last;
                }
                warn "not all items removed from busy queue ($count)\n"
                    if $self->warn_on_requeue && $count != @done;

                # put back the claimed but not touched items
                if (@items > 0) {
                    my $n = @items;
                    print "unclaiming $n items\n";
                    for (1 .. $MAX_RECONNECT) {
                        eval { $self->unclaim($_) for @items; 1; }
                        or do {
                            last if $stop;
                            sleep 1;
                            next;
                        };
                        last;
                    }
                }
                $stop = 1 if ($maxitems > 0 && ($maxitems -= @done) <= 0)
                                || ($stop_time > 0 && time() >= $stop_time);
            }
        }
    } # end 'sub consume'
} # end SCOPE

sub _clean_error {
    $_[0] =~ s/,  line [0-9]+//;
    chomp $_[0];
    return $_[0];
}

# methods to be used for cleanup script and Nagios checks
# the methods read or remove items from the busy queue
sub handle_expired_items {
    my ($self, $timeout, $action) = @_;
    $timeout ||= 10;
    die "timeout should be a number> 0" if not int($timeout);
    die "unknown action"
        if not $action or $action !~ /^(?:requeue|drop)$/;
    my $conn = $self->redis_conn;
    my @serial = $conn->lrange($self->_busy_queue, 0, -1);
    my $time = time;
    my %timetable =
        map { reverse split /-/,$_,2 }
        $conn->lrange($self->_time_queue, 0, -1);
    my @match = grep { exists $timetable{$_} } @serial;
    my %match = map { $_ => undef } @match;
    my @timedout = grep { $time - $timetable{$_} >= $timeout } @match;
    my @log;

    if ($action eq 'requeue') {
        for my $serial (@timedout) {
            my $item = Queue::Q::ReliableFIFO::Item->new(
                _serialized => $serial
            );
            my $n = $self->requeue_busy_item($item);
            push @log, $item
                if $n;
        }
    }
    elsif ($action eq 'drop') {
        for my $serial (@timedout) {
            my $n = $conn->lrem( $self->_busy_queue, -1, $serial);
            push @log, Queue::Q::ReliableFIFO::Item->new(_serialized => $serial)
                if $n;
        }
    }

    # We create a new timetable. We take the original timetable and
    # exclude:
    # 1. the busy items which timed out and we just handled
    # 2. timetable items which have no corresponding busy items anymore
    my %timedout = map { $_ => undef } @timedout;
    my %busy = map { $_ => undef } @serial;
    my %newtimetable =
        map  { $_ => $timetable{$_} }
        grep { exists $busy{$_} }        # exclude (ad 2.)
        grep { ! exists $timedout{$_} }  # exclude (ad 1.)
        keys %timetable;                 # original timetable
    # put in the items of latest scan we did not see before
    $newtimetable{$_} = $time
        for (grep { !exists $newtimetable{$_} } @serial);
    $conn->multi;
    $conn->del($self->_time_queue);
    $conn->lpush($self->_time_queue, join('-',$newtimetable{$_},$_))
        for (keys %newtimetable);
    $conn->exec;
    #FIXME the log info should also show what is done with the items
    # (e.g. dropped after requeue limit).
    return @log;
}
1;

__END__