File Coverage

blib/lib/Minion/Backend/Storable.pm

Criterion	Covered	Total	%
statement	203	203	100.0
branch	59	66	89.3
condition	87	100	87.0
subroutine	42	42	100.0
pod	18	18	100.0
total	409	429	95.3

package Minion::Backend::Storable;
use Minion::Backend -base;

our $VERSION = 6.051;

use Sys::Hostname 'hostname';
use Time::HiRes qw(time usleep);

# Attributes

has 'file';

# Methods

sub broadcast {
  my ($self, $command, $args, $ids) = (shift, shift, shift || [], shift || []);

  my $guard = $self->_guard->_write;
  my $inboxes = $guard->_inboxes;
  my $workers = $guard->_workers;
  @$ids = @$ids ? map exists($workers->{$_}), @$ids
      : keys %$workers unless @$ids;

  push @{$inboxes->{$_} ||= []}, [$command, @$args] for @$ids;

  return !!@$ids;
}

sub dequeue {
  my ($self, $id, $wait, $options) = @_;
  usleep $wait * 1_000_000 unless my $job = $self->_try($id, $options);
  return $job || $self->_try($id, $options);
}

sub enqueue {
  my ($self, $task, $args, $options) = (shift, shift, shift || [], shift || {});

  my $guard = $self->_guard->_write;

  my $job = {
    args     => $args,
    attempts => $options->{attempts} // 1,
    created  => time,
    delayed  => time + ($options->{delay} // 0),
    id       => $guard->_job_id,
    parents  => $options->{parents} || [],
    priority => $options->{priority} // 0,
    queue    => $options->{queue} // 'default',
    retries  => 0,
    state    => 'inactive',
    task     => $task
  };
  $guard->_jobs->{$job->{id}} = $job;

  return $job->{id};
}

sub fail_job { shift->_update(1, @_) }

sub finish_job { shift->_update(0, @_) }

sub job_info {
  my ($self, $id) = @_;
  my $guard = $self->_guard;
  my $jobs = $guard->_jobs;
  my $job = $jobs->{$id} or return undef;
  $job->{children} = $guard->_children($id);
  return $job;
}

sub list_jobs {
  my ($self, $offset, $limit, $options) = @_;

  my $guard = $self->_guard;
  my @jobs = sort { $b->{created} <=> $a->{created} }
  grep +( (not exists $options->{queue} or $_->{queue} eq $options->{queue})
      and (not exists $options->{state} or $_->{state} eq $options->{state})
      and (not exists $options->{task}  or $_->{task} eq $options->{task})
  ), values %{$guard->_jobs};

  return [map +($_->{children} = $guard->_children($_->{id}) and $_), grep defined, @jobs[$offset .. ($offset + $limit - 1)]];
}

sub list_workers {
  my ($self, $offset, $limit) = @_;
  my $guard = $self->_guard;
  my @workers = map { $self->_worker_info($guard, $_->{id}) }
    sort { $b->{started} <=> $a->{started} } values %{$guard->_workers};
  return [grep {defined} @workers[$offset .. ($offset + $limit - 1)]];
}

sub new { shift->SUPER::new(file => shift) }

sub receive {
  my ($self, $id) = @_;
  my $guard = $self->_guard->_write;
  my $inboxes = $guard->_inboxes;
  my $inbox = $inboxes->{$id} || [];
  $inboxes->{$id} = [];
  return $inbox;
}

sub register_worker {
  my ($self, $id, $options) = (shift, shift, shift || {});
  my $guard = $self->_guard->_write;
  my $worker = $id ? $guard->_workers->{$id} : undef;
  unless ($worker) {
    $worker = {host => hostname, id => $guard->_id, pid => $$, started => time};
    $guard->_workers->{$worker->{id}} = $worker;
  }
  @$worker{qw(notified status)} = (time, $options->{status} || {});
  return $worker->{id};
}

sub remove_job {
  my ($self, $id) = @_;
  my $guard = $self->_guard;
  delete $guard->_write->_jobs->{$id}
    if my $removed = !!$guard->_job($id, qw(failed finished inactive));
  return $removed;
}

sub repair {
  my $self = shift;
  my $minion = $self->minion;

  # Workers without heartbeat
  my $guard   = $self->_guard->_write;
  my $workers = $guard->_workers;
  my $jobs    = $guard->_jobs;
  my $after   = time - $minion->missing_after;
  $_->{notified} < $after and delete $workers->{$_->{id}} for values %$workers;

  # Jobs with missing parents (cannot be retried)
  for my $job (values %$jobs) {
    next unless $job->{parents} and $job->{state} eq 'inactive';
    my $missing;
    for my $p (@{$job->{parents}}) {
      ++$missing and last unless exists $jobs->{$job->{id}};
    }
    @$job{qw(finished result state)} = (time, 'Parent went away', 'failed');
  }

  # Old jobs without unfinished dependents
  $after = time - $minion->remove_after;
  for my $job (values %$jobs) {
    next unless $job->{state} eq 'finished' and $job->{finished} <= $after;
    my $id = $job->{id};
    delete $jobs->{$id} unless grep +($jobs->{$_}{state} ne 'finished'),
        @{$guard->_children($id)};
  }

  # Jobs with missing worker (can be retried)
  my @abandoned = map [@$_{qw(id retries)}],
      grep +($_->{state} eq 'active' and not exists $workers->{$_->{worker}}),
      values %$jobs;
  undef $guard;
  $self->fail_job(@$_, 'Worker went away') for @abandoned;

  return;
}

sub reset { shift->_guard->_spurt({}) }

sub retry_job {
  my ($self, $id, $retries, $options) = (shift, shift, shift, shift || {});

  my $guard = $self->_guard;
  return undef
    unless my $job = $guard->_job($id, qw(active failed finished inactive));
  return undef unless $job->{retries} == $retries;
  $guard->_write;
  ++$job->{retries};
  $job->{delayed} = time + $options->{delay} if $options->{delay};
  exists $options->{$_} and $job->{$_} = $options->{$_} for qw(priority queue);
  @$job{qw(retried state)} = (time, 'inactive');
  delete @$job{qw(finished started worker)};

  return 1;
}

sub stats {
  my $self = shift;

  my ($active, $delayed) = (0, 0);
  my (%seen, %states);
  my $guard = $self->_guard;
  for my $job (values %{$guard->_jobs}) {
    ++$states{$job->{state}};
    ++$active if $job->{state} eq 'active' and not $seen{$job->{worker}}++;
    ++$delayed if $job->{state} eq 'inactive'
        and (time < $job->{delayed} or @{$job->{parents}});
  }

  return {
    active_workers   => $active,
    inactive_workers => keys(%{$guard->_workers}) - $active,
    active_jobs      => $states{active}   // 0,
    delayed_jobs     => $delayed,
    enqueued_jobs    => $guard->_job_count,
    failed_jobs      => $states{failed}   // 0,
    finished_jobs    => $states{finished} // 0,
    inactive_jobs    => $states{inactive} // 0
  };
}

sub unregister_worker {
  my ($self, $id) = @_;
  my $guard = $self->_guard->_write;
  delete $guard->_inboxes->{$id};
  delete $guard->_workers->{$id};
}

sub worker_info { $_[0]->_worker_info($_[0]->_guard, $_[1]) }

sub _guard { Minion::Backend::Storable::_Guard->new(backend => shift) }

sub _try {
  my ($self, $id, $options) = @_;
  my $tasks = $self->minion->tasks;
  my %queues = map +($_ => 1), @{$options->{queues} || ['default']};

  my $now = time;
  my $guard = $self->_guard;
  my $jobs = $guard->_jobs;
  my @ready = sort {
      $b->{priority} <=> $a->{priority} || $a->{created} <=> $b->{created} }
    grep +($_->{state} eq 'inactive' and $queues{$_->{queue}}
      and $tasks->{$_->{task}} and $_->{delayed} < $now),
    values %{$jobs};

  my $job;
  CANDIDATE: for my $candidate (@ready) {
    $job = $candidate and last
      unless my @parents = @{$candidate->{parents} || []};
    ($jobs->{$_}{state} // '') ne 'finished' and next CANDIDATE for @parents;
    $job = $candidate;
  }
  return undef unless $job;
  $guard->_write;
  @$job{qw(started state worker)} = (time, 'active', $id);
  return $job;
}

sub _update {
  my ($self, $fail, $id, $retries, $result) = @_;

  my $guard = $self->_guard;
  return undef unless my $job = $guard->_job($id, 'active');
  return undef unless $job->{retries} == $retries;

  $guard->_write;
  @$job{qw(finished result)} = (time, $result);
  $job->{state} = $fail ? 'failed' : 'finished';
  undef $guard;

  return 1 unless $fail and $job->{attempts} > $retries + 1;
  my $delay = $self->minion->backoff->($retries);
  return $self->retry_job($id, $retries, {delay => $delay});
}

sub _worker_info {
  my ($self, $guard, $id) = @_;

  return undef unless $id && (my $worker = $guard->_workers->{$id});
  my @jobs = map $_->{id},
      grep +($_->{state} eq 'active' and $_->{worker} eq $id),
      values %{$guard->_jobs};

  return {%$worker, jobs => \@jobs};
}

package Minion::Backend::Storable::_Guard;
use Mojo::Base -base;

use Fcntl ':flock';
use Digest::MD5 'md5_hex';
use Storable qw(retrieve store);

sub DESTROY {
  my $self = shift;
  $self->_spurt($self->_data) if $self->{write};
  flock $self->{lock}, LOCK_UN;
}

sub new {
  my $self = shift->SUPER::new(@_);
  $self->_spurt({}) unless -f (my $file = $self->{backend}->file);
  open $self->{lock}, '>', "$file.lock";
  flock $self->{lock}, LOCK_EX;
  return $self;
}

sub _children {
  my ($self, $id) = @_;
  my $children = [];
  for my $job (values %{$self->_jobs}) {
    push @$children, $job->{id} if grep +($_ eq $id), @{$job->{parents} || []};
  }
  return $children;
}

sub _data { $_[0]{data} ||= retrieve($_[0]{backend}->file) }

sub _id {
  my $self = shift;
  my $id;
  do { $id = md5_hex(time . rand 999) } while $self->_workers->{$id};
  return $id;
}

sub _inboxes { shift->_data->{inboxes} ||= {} }

sub _job {
  my ($self, $id) = (shift, shift);
  return undef unless my $job = $self->_jobs->{$id};
  return(grep(($job->{state} eq $_), @_) ? $job : undef);
}

sub _job_id {
  my $self = shift;
  my $id;
  do { $id = md5_hex(time . rand 999) } while $self->_jobs->{$id};
  ++$self->_data->{job_count};
  return $id;
}

sub _job_count { shift->_data->{job_count} //= 0 }

sub _jobs { shift->_data->{jobs} ||= {} }

sub _spurt { store($_[1] => $_[0]{backend}->file) }

sub _workers { shift->_data->{workers} ||= {} }

sub _write { ++$_[0]{write} && return $_[0] }

1;
__END__