view release on metacpan or  search on metacpan

lib/Convert/Pheno/HTTP/Jobs.pm  view on Meta::CPAN

package Convert::Pheno::HTTP::Jobs;

use strict;
use warnings;
use Cwd qw(abs_path);
use Digest::SHA qw(sha256_hex);
use Encode qw(decode FB_DEFAULT);
use File::Copy qw(copy);
use File::Find qw(find);
use File::Path qw(make_path remove_tree);
use File::Spec;
use Fcntl qw(LOCK_EX LOCK_SH LOCK_NB);
use IPC::Open3 qw(open3);
use JSON::XS;
use Mojo::IOLoop;
use Mojo::Util qw(sha1_sum);
use Path::Tiny qw(path);
use POSIX qw(WNOHANG);
use Symbol qw(gensym);
use Convert::Pheno::HTTP::Service qw(execute execute_files catalog);
use Convert::Pheno::IO::Atomic qw(write_atomically);

my $JSON = JSON::XS->new->utf8->canonical->pretty;
my $MAX_CONCURRENT_JOBS = 16;

sub _validate_job_limit {
    my ($value, $maximum) = @_;
    $maximum //= $MAX_CONCURRENT_JOBS;
    die "Maximum concurrent jobs must be an integer from 1 to $maximum\n"
      unless defined($value) && !ref($value) && "$value" =~ /\A[1-9][0-9]*\z/
      && $value <= $maximum;
    return 0 + $value;
}

sub _metadata_lock {
    my ($file, $mode) = @_;
    # Lock a stable sibling, not the JSON inode replaced by atomic writes.
    # Windows readers must not overlap replacement (including its backup gap).
    # Keep this file in place: unlinking it could create two independent locks.
    my $lockfile = path($file)->parent->child('.metadata.lock');
    open my $lock, '>>', $lockfile
      or die "Cannot open job metadata lock <$lockfile>: $!\n";
    flock($lock, $mode)
      or die "Cannot lock job metadata <$lockfile>: $!\n";
    return $lock;
}

sub _write {
    my ($file, $data) = @_;
    my $lock = _metadata_lock($file, LOCK_EX);
    write_atomically($file, sub { path($_[0])->spew_raw($JSON->encode($data)) });
}
sub _read {
    my ($file) = @_;
    my $lock = _metadata_lock($file, LOCK_SH);
    return $JSON->decode(path($file)->slurp_raw);
}
sub _id { sha1_sum(join ':', $$, rand(), time(), {}) }
sub _resolved_path {
    my ($value) = @_;
    return unless defined $value;
    $value = "$value";
    my $resolved = abs_path($value);
    if (!defined $resolved) {
        my $candidate = path($value);
        my $parent = abs_path("@{[$candidate->parent]}");
        $resolved = defined($parent)
          ? File::Spec->catfile($parent, $candidate->basename)
          : File::Spec->rel2abs($value);
    }
    return File::Spec->canonpath($resolved);
}
sub _path_key {
    my ($value) = @_;
    my $key = _resolved_path($value);
    return unless defined $key;
    if ($^O eq 'MSWin32') {
        $key =~ tr{\\}{/};
        $key = lc $key;
    }
    return $key;
}
sub _same_path {
    my ($left, $right) = @_;
    return defined($left) && defined($right) && _path_key($left) eq _path_key($right);
}
sub _worker_lock {
    my ($dir) = @_;
    open my $fh, '>>', path($dir, '.worker.lock') or die "Cannot lock run folder\n";
    flock($fh, LOCK_EX | LOCK_NB) or die "A previous conversion is still finishing; wait before reopening this workspace\n";
    return $fh;
}

sub _recover_publication {
    my ($dir, $status) = @_;
    my $journal = path($dir, 'publication.json');
    return $status unless -f $journal;
    my $publication = _read($journal);