flac test threaded
Posted by Anonymous on Mon 17th Sep 2012 21:59
raw | new post

#!/usr/bin/perl

use 5.14.0;
use strict;
use warnings;
use Cwd qw(abs_path);

use threads qw(yield);
use threads::shared;
use Thread::Queue;
use Thread::Semaphore;

my $clear = `clear && echo`;
chomp(my $cores = `grep -c ^processor /proc/cpuinfo`);
my $library = "/home/sir/Music/Songbird Music";
my $logf = "$ENV{HOME}/flac_test.log";
my $n :shared = 0;
my @err :shared;
my $end :shared;

my %flac :shared;
my %done :shared;
my $total :shared = 0;
my $disk_size = 1000000000;

# This will be used to control access to the logger subroutine.
my $semaphore = Thread::Semaphore->new();
my $semaphore2 = Thread::Semaphore->new();
my $semaphore3 = Thread::Semaphore->new();

# Creating the thread queue.
my $q = Thread::Queue->new();

open(my $LOG, '>>', $logf) or die "Can\'t open '$logf': $!"; # Open file handle for the log file
$LOG->autoflush(1);     # Make the $LOG file handle unbuffered for instant logging.
open(my $SE, ">&STDERR") or die "Can\'t duplicate STDERR: $!"; # Duplicate STDERR as a regular file handle

#sub msay {
#       my @fh;
#       push(@fh, shift) while (ref($_[0]) eq 'GLOB');
#       say $_ @_ foreach (@fh);
#}

sub getdirs {

        my $dn = $library;
        my @dirs;

        open(FIND, '-|', qq{find -L "$dn" -name "*" -type d})
        or die "Can't run 'find': $!";
        push(@dirs, (<FIND>));
        chomp(@dirs);
        close(FIND) or die "Can't close 'find': $!";
        return(@dirs);
}

sub getfiles {
        my $dn = shift;
        my (@files, $fn);
        opendir(my $dh, $dn) or die "Can\'t open directory '$_': $!";
        foreach (readdir($dh)) {
                if (/.flac$/) {
                        $fn = $dn . '/' . $_;
                        push(@files, $fn) if -f $fn;
                }
        }
        closedir $dh or die "Can\'t close directory '$dn': $!";
        return @files;
}

#### Subroutine for controlling the log file
sub logger {

        $semaphore->down();

        my($arg, $sw, $fn, $n);
        my $now = localtime(time);
        my @larg = qw{start int change end}; # Array of accepted switches to this subroutine

        #### Loop through all the arguments passed to this subroutine
        #### Perform checks that decide which variable the arguments are to be assigned to
        while (@_) {

                        $arg = shift(@_);

                        foreach (@larg) {
                                if ($_ eq $arg) { $sw = $arg }
                        }
                        #### If $arg is a number assign it to $n, if it's a file assign it to $fn
                        if ($arg =~ /^[0-9]*$/) { $n = $arg; }
                        else { $fn = $arg; }
        }

        given ($sw) {
                #### Starts writing the log
                when ('start') {
                        say $LOG "\n**** Logging started on $now ****\n";
                }
                #### Called when file has been changed
                when ('change') {
                        say $LOG $fn . "\n\t" . "has been corrupted.\n";
                        lock(@err);
                        push(@err, $fn);
                }
                #### Called when done, and to close the log
                #### If no errors occured write "Everything is OK!" to the log
                #### If errors occurred print the @err array
                #### Either way, print number of files processed
                when ('end') {
                        if (!@err) {
                                say $LOG "\nEverything is OK!\n";
                        } else {
                                foreach my $fn (@err) {
                                        say $SE "ERROR: " . $fn;
                                }
                        }

                        say $LOG $n . " file(s) were tested.\n" if ($n);
                        say $LOG "\n**** Logging ended on $now ****\n";
                        close $LOG or die "Can\'t close '$LOG': $!";
                }
        }
        $semaphore->up();
}

sub ram {
                my $fn = shift;
                my $size = (stat($fn))[7];
                open(FILE, '<:raw', $fn) or die "Can't open '$fn': $!";
                read(FILE, $flac{$fn}, $size);
                close(FILE) or die "Can't close '$fn': $!";
                $semaphore2->down();
                $done{$fn} = 1;
                $total += length($flac{$fn});
                $semaphore2->up();
}

sub test {

        while (!$q->pending() && !$end) {
                yield();
        }

        my $tid = threads->tid();

        LOOP: while (my $fn = $q->dequeue_nb()) {

                say "thread${tid} $fn: testing...";
                my $pid = open(FLAC, '|-:raw', qq{flac --totally-silent --test -})
                or die "can't run 'flac': $!";
                print FLAC $flac{$fn};
                close(FLAC);

                my $status = ${^CHILD_ERROR_NATIVE};

                if ($status != 0 && $status != 2) {
                        say $pid;
                        kill 9, $pid;
                        close(FLAC) or die "Can't close 'flac': $!";
                        logger('change', $fn);
                }

                $semaphore3->down();
                $total -= length($flac{$fn});
                delete($flac{$fn});
                $n++;
                $semaphore3->up();
        }

        if (!$q->pending() && !$end) {
                say "thread${tid}: yielding!";
                yield();
                goto(LOOP);
        }
}

say "Starting threads";
my @threads;
foreach (1 .. $cores) {
        push(@threads, threads->create(\&test));
}

logger('start');


foreach my $dn (getdirs) {
        my @files = getfiles($dn);

        foreach my $fn (@files) {
                my $running_threads = threads->list(threads::running);
                my $pending = $q->pending();
                say "$running_threads running threads : $pending pending files : $total bytes in RAM";

                ram($fn);

                # If the RAM disk is full, put decoded files in the queue,
                # and then yield until the LAME threads have cleared some RAM.

                while ($total > $disk_size) {
                        foreach my $fn (keys(%done)) {
                                $q->enqueue($fn);
                                lock(%done);
                                delete($done{$fn});
                        }
                        yield();
                }
        }

        foreach my $fn (keys(%done)) {
                $q->enqueue($fn);
                lock(%done);
                delete($done{$fn});
        }
}

$end = 1;
foreach my $t (threads->list()) { $t->join(); }
say("All threads joined");
logger('end', $n);