File Coverage

blib/lib/Coro/PrioChannel.pm

Criterion	Covered	Total	%
statement	7	9	77.7
branch			n/a
condition			n/a
subroutine	3	3	100.0
pod			n/a
total	10	12	83.3

package Coro::PrioChannel;
{
  $Coro::PrioChannel::VERSION = '0.005';
}
use strict;
use warnings;

# ABSTRACT: Priority message queues for Coro


use Coro qw(:prio);
use Coro::Semaphore ();

use List::Util qw(first sum);
use AnyEvent ();

sub SGET()      { 0 }
sub SPUT()      { 1 }
sub REPRIO()    { 2 }
sub NEXTCHECK() { 3 }
sub DATA()      { 4 }
sub MAX()       { PRIO_MAX - PRIO_MIN + DATA + 1 }


sub new {
   # we cheat, just like Coro::Channel.
   bless [
      (Coro::Semaphore::_alloc 0), # counts data
      (Coro::Semaphore::_alloc +($_[1] || 2_000_000_000) - 1), # counts remaining space
      $_[2], # reprioritization check time
      (defined $_[2] ? (AnyEvent->now + $_[2]) : undef), # last reprioritization check
      [], # initially empty
   ]
}


sub _put {
   my $after = (AnyEvent->now + $_[0]->[REPRIO]) if defined $_[0]->[REPRIO];
   push @{$_[0][DATA + ($_[2]||PRIO_NORMAL) - PRIO_MIN()]}, [$_[1], $after];
}

sub put {
   $_[0]->reprioritize;
   _put @_;
   Coro::Semaphore::up   $_[0][SGET];
   Coro::Semaphore::down $_[0][SPUT];
}


sub get {
   Coro::Semaphore::down $_[0][SGET];
   Coro::Semaphore::up   $_[0][SPUT];

   my $a = first { $_ && scalar @$_ } reverse @{$_[0]}[DATA..MAX];

   ref $a ? shift(@$a)->[0] : undef;
}


sub reprioritize {
    return unless defined $_[0]->[REPRIO];

    my $now = AnyEvent->now;
    return unless $_[0]->[NEXTCHECK] <= $now;

    my $q = $_[0];
    foreach my $pri (PRIO_MIN .. PRIO_HIGH) {
        my $next_pri = $pri + 1;
        my $idx      = DATA + $pri - PRIO_MIN;
        my @keep;

        foreach my $item (@{$q->[$idx]}) {
            if ($item->[1] <= $now) {
                _put $q, $item->[0], $next_pri;
            } else {
                push @keep, $item;
            }
        }

        $q->[$idx] = \@keep;
    }

    $q->[NEXTCHECK] = $now + $q->[REPRIO];
    return;
}


sub shutdown {
   Coro::Semaphore::adjust $_[0][SGET], 1_000_000_000;
}


sub size {
    my $min = @_ > 1 ? $_[1] - PRIO_MIN + DATA : DATA;
    sum map { $_ ? scalar @$_ : 0 } @{$_[0]}[$min..MAX];
}


1;

__END__