A better way to convert your mp3 tags using Perl, POE, Linux::Inotify2 and POE::Wheel::Run (for you Sony PS3 or other media player)

So, what we have here is a country that tries to run itself on the commandments of a god who the people feel maybe wearing his underpants on his head. Has he abominated underpants?

No sir, but is probably only a matter of time.

Yesterday, I posted Howto: Convert your mp3 tags (id3v2 to id3v1) so your Playstation 3 can play your MP3s! and it worked fine but there was one little problem with it.

When we processed the files, namely running eye3D, we did so synchronously. Essentially, the file notification came in from the Linux kernel and we processed the file at that time. This may be an issue of overflowing the inotify queue within the Linux kernel if there are a lot of files to process.

A better solution would be to add the file to an internal queue and process the files in a sub process using POE::Wheel::Run. Of course we will limit the number of sub processes 🙂

#!/usr/bin/perl

use strict;
use warnings;

use File::Basename;
use File::Find ();
use Getopt::Std;
use Linux::Inotify2;
use POE qw( Kernel Session Wheel::Run );

$|++;

#######################################
#######################################

our @found_dirs;
our $max_concurrent_tasks;

sub watch_add_dir {
my ($heap_ref, $session, $dir_name) = @_;

##############
# Watch this directory with a call back
# to the watch_hdlr() subroutine via
# a message to the POE system
##############
$heap_ref->{inotify}->watch($dir_name, IN_CREATE|IN_CLOSE_WRITE, $session->postback("watch_hdlr"));
print " Watching directory $dir_name\n";
}

sub watch_hdlr {
my ($kernel, $heap, $session, $event) = ( $_[KERNEL], $_[HEAP], $_[SESSION], $_[ARG1][0] );

my $name = $event->fullname;
my $short_name = $event->name;

##############
# We can receive many many notifications
# for a file. If we've already processed
# the file, do nothing.
##############
unless ($heap->{inotify}{files}{$name}) {

##############
# If a new directory is added, we need
# to watch that directory too.
##############
if ($event->IN_CREATE && -d $name) {
print "New directory: $name\n";
watch_add_dir($heap, $session, $name);
} elsif ($event->IN_CLOSE_WRITE) {

##############
# When a file descriptor that was opened for
# 'writing' is closed, then process that
# file it was being written to. We're
# assuming that the file is complete at this
# point as the operation will be a copy into
# the watched directory
##############
my $ext = ( fileparse($name, '\..*') )[2];

if (lc($ext) eq '.mp3') {
##############
# Add the file to the file process queue
##############
push @{ $heap->{task}{task_files} }, $name;

##############
# Mark that we have processed the file. If
# we don't we will end up processing the file
# in an infinite loop because we are modifying
# the files.
##############
$heap->{inotify}{files}{$name} = 1;

##############
# Yield to "task_next_file" through so
# that we can process files in the queue.
##############
$kernel->yield("task_next_file");
}

$heap->{inotify}{files}{$name} = 1;
}
}

##############
# While possible, it is highly unlikely that we will
# overflow the notification buffers within the Linux
# kernel. If so, we should report that.
##############
print "events for $name have been lost\n" if $event->IN_Q_OVERFLOW;
}

sub task_next_file {
my ($kernel, $heap) = @_[ KERNEL, HEAP ];

##############
# Process the files in the queue up
# to the $max_concurrent_tasks at
# once. Any extras will be processed
# when a file (task) completes.
##############
while ( keys( %{ $heap->{task} } ) < $max_concurrent_tasks ) {
my $next_task_file = shift @{ $heap->{task}{task_files} };

##############
# If the $next_task_file is empty, then we can safely
# ignore it.
##############
last unless defined $next_task_file;

##############
# Use POE::Wheel::Run to fire off the
# file processing using a sub process
# to the process_file() subroutine
##############
my $task = POE::Wheel::Run->new (
Program => sub { process_file($next_task_file) },
StdoutEvent => "task_output",
CloseEvent => "task_done",
);

##############
# Update the session with the task
# information and the kernel with
# the SIG_CHILD handler. These are
# necessary for the task to execute.
##############
$heap->{task}->{ $task->ID } = $task;
$kernel->sig_child( $task->PID, "sig_child");
}
}

sub process_file {
my $file = shift;

print " Processed \"$file\"\n";

##############
# Use the eyeD3 package to convert
# the mp3 id3v2/3/4 to id3v1. If
# eyeD3 fails, we don't really care. 🙂
##############
my $cmd_output = `eyeD3 --to-v1.1 "$file"`;
$cmd_output = `eyeD3 --remove-v2 "$file"`;
}

sub find_wanted {
my $object = $File::Find::name;

if (-d $object) {
push @found_dirs, $object;
}
}

#######################################
#######################################
#######################################

my %arg_options;
my $watch_dir;

getopts('d:t:', \%arg_options);

if ($arg_options{d} && -d $arg_options{d}) {
$watch_dir = $arg_options{d};

if ($arg_options{t} && $arg_options{t} =~ /^\d+/) {
$max_concurrent_tasks = $arg_options{t};
} else {
$max_concurrent_tasks = 2;
}

##############
# We need to watch all existing sub directories
# so we will find them and add them to the
# @found_dirs array to be added to the watched
# directories when we create the Inotify object
##############
File::Find::find({wanted => \&find_wanted}, $watch_dir);

POE::Session->create
( inline_states =>
{ _start => sub {
my $inotify_FH;

##############
# alias this particular POE session to
# 'notify' so we can easily reference
# it later if needed
##############
$_[KERNEL]->alias_set('notify');

##############
# Create the Linux::INotify object
##############
$_[HEAP]{inotify} = new Linux::Inotify2
or die "Unable to create new inotify object: $!";

##############
# Add the preexisting directories to
# be watched from the @found_dirs array
##############
foreach my $dir (@found_dirs) {
watch_add_dir($_[HEAP], $_[SESSION], $dir);
}

##############
# We need to create a hash in the "notify"
# POE session so we can determine if we've
# processed a file already
##############
$_[HEAP]{inotify}{files} = {};

##############
# The Inotify notifications are received
# on a file descriptor. We need to read
# from it when there is something to be
# read
##############
open $inotify_FH, "< &=" . $_[HEAP]{inotify}->fileno
or die "Can’t fdopen: $!\n";

##############
# Inform POE to poll the file descriptor
##############
$_[KERNEL]->select_read( $inotify_FH, "inotify_poll" );
},
inotify_poll => sub {
$_[HEAP]{inotify}->poll;
},
watch_hdlr => \&watch_hdlr,

##############
# Process the next file in the queue
##############
task_next_file => \&task_next_file,

##############
# print the output of the job
##############
task_output => sub {
my $result = $_[ARG0];

print "$result\n";
},

##############
# When we are done with a file, go process the
# next file if there is one waiting
##############
task_done => sub {
my ($kernel, $heap, $task_id) = @_[ KERNEL, HEAP, ARG0 ];

delete $heap->{task}{$task_id};
$kernel->yield("task_next_file");
},
sig_child => sub {
my ($heap, $pid) = @_[ HEAP, ARG1 ];

my $details = delete $heap->{$pid};
},
},
);

POE::Kernel->run();
}

Example output:

ps3_mp3_converter.pl -d /home/jfroebe/j
 Watching directory /home/jfroebe/j
Watching directory /home/jfroebe/j/bin
Watching directory /home/jfroebe/j/doc
Watching directory /home/jfroebe/j/java
Watching directory /home/jfroebe/j/lib
Watching directory /home/jfroebe/j/j
Watching directory /home/jfroebe/j/j/tmp
Watching directory /home/jfroebe/j/sdk
Watching directory /home/jfroebe/j/sdk/demo
Watching directory /home/jfroebe/j/sdk/include
New directory: /home/jfroebe/j/Earth Final Conflict Soundtrack
Watching directory /home/jfroebe/j/Earth Final Conflict Soundtrack
Processed "/home/jfroebe/j/Earth Final Conflict Soundtrack/01 Main Title.mp3"
Processed "/home/jfroebe/j/Earth Final Conflict Soundtrack/02 The Scret of Strandhill-Redemption.mp3"
Processed "/home/jfroebe/j/Earth Final Conflict Soundtrack/03 Old Flame.mp3"
Processed "/home/jfroebe/j/Earth Final Conflict Soundtrack/04 Defector.mp3"
Processed "/home/jfroebe/j/Earth Final Conflict Soundtrack/05 Decidion.mp3"
Processed "/home/jfroebe/j/Earth Final Conflict Soundtrack/06 Float Like a Butterfly.mp3"
Processed "/home/jfroebe/j/Earth Final Conflict Soundtrack/07 Sandoval's Run.mp3"
Processed "/home/jfroebe/j/Earth Final Conflict Soundtrack/08 Bliss.mp3"
Processed "/home/jfroebe/j/Earth Final Conflict Soundtrack/09 If You Could Read My Mind.mp3"
Processed "/home/jfroebe/j/Earth Final Conflict Soundtrack/10 Lilli.mp3"
Processed "/home/jfroebe/j/Earth Final Conflict Soundtrack/11 Law and Order.mp3"
Processed "/home/jfroebe/j/Earth Final Conflict Soundtrack/12 Atavus.mp3"
Processed "/home/jfroebe/j/Earth Final Conflict Soundtrack/13 Between Heaven and Hell.mp3"
Processed "/home/jfroebe/j/Earth Final Conflict Soundtrack/14 Sleepers.mp3"
Processed "/home/jfroebe/j/Earth Final Conflict Soundtrack/15 Dimensions.mp3"
Processed "/home/jfroebe/j/Earth Final Conflict Soundtrack/16 Moonscape.mp3"
Processed "/home/jfroebe/j/Earth Final Conflict Soundtrack/17 Isabel.mp3"
Processed "/home/jfroebe/j/Earth Final Conflict Soundtrack/18 The Gauntlet.mp3"
Processed "/home/jfroebe/j/Earth Final Conflict Soundtrack/19 Second Chances.mp3"
Processed "/home/jfroebe/j/Earth Final Conflict Soundtrack/20 One Man's Castle.mp3"
Processed "/home/jfroebe/j/Earth Final Conflict Soundtrack/21 Payback.mp3"
Processed "/home/jfroebe/j/Earth Final Conflict Soundtrack/22 Truth.mp3"
Processed "/home/jfroebe/j/Earth Final Conflict Soundtrack/23 Déjà Vu.mp3"
Processed "/home/jfroebe/j/Earth Final Conflict Soundtrack/24 Crossfire.mp3"
Processed "/home/jfroebe/j/Earth Final Conflict Soundtrack/25 Volunteers-End Credits.mp3"
Share Button

Howto: Convert your mp3 tags (id3v2 to id3v1) so your Playstation 3 can play your MP3s!

  1. Run the converter on your media server: ps3_mp3_converter.pl -d {directory}
  2. Copy your mp3 collection wherever you told ps3_mp3_converter.pl to run in.
#!/usr/bin/perl

use strict;
use warnings;

use File::Basename;
use File::Find ();
use Getopt::Std;
use Linux::Inotify2;
use POE;

$|++;

#######################################
#######################################

our @found_dirs;

sub watch_add_dir {
 my ($heap_ref, $session, $dir_name) = @_;

 $heap_ref->{inotify}->watch($dir_name, IN_CREATE|IN_CLOSE_WRITE, $session->postback("watch_hdlr"));
 print " Watching directory $dir_name\n";
}

sub watch_hdlr {
 my ($heap_ref, $session, $event) = ( $_[HEAP], $_[SESSION], $_[ARG1][0] );

 my $name = $event->fullname;
 my $short_name = $event->name;

 unless ($_[HEAP]{inotify}{files}{$name}) {
  if ($event->IN_CREATE && -d $name) {
   print "New directory: $name\n";
   watch_add_dir($heap_ref, $session, $name);
  } elsif ($event->IN_CLOSE_WRITE) {
   my $ext = ( fileparse($name, '\..*') )[2];

   if (lc($ext) eq '.mp3') {
    print "-"x20 . "\n";
    print "$name:\n";

    my $cmd_output = `eyeD3 --to-v1.1 "$name"`;
    $cmd_output = `eyeD3 --remove-v2 "$name"`;
   }

   $_[HEAP]{inotify}{files}{$name} = 1;
  }
 }

 print "events for $name have been lost\n" if $event->IN_Q_OVERFLOW;
}

sub find_wanted {
 my $object = $File::Find::name;

 if (-d $object) {
  push @found_dirs, $object;
 }
}

#######################################
#######################################
#######################################

my %arg_options;
my $watch_dir;

getopts('d:', \%arg_options);

if ($arg_options{d} && -d $arg_options{d}) {
 $watch_dir = $arg_options{d};
 File::Find::find({wanted => \&find_wanted}, $watch_dir);

 POE::Session->create
  ( inline_states =>
   { _start => sub {
     my $inotify_FH;

     $_[KERNEL]->alias_set('notify');
     $_[HEAP]{inotify} = new Linux::Inotify2
      or die "Unable to create new inotify object: $!";

     foreach my $dir (@found_dirs) {
      watch_add_dir($_[HEAP], $_[SESSION], $dir);
     }

     $_[HEAP]{inotify}{files} = {};

     open $inotify_FH, "< &=" . $_&#91;HEAP&#93;{inotify}->fileno
     or die "Can’t fdopen: $!\n";

     $_[KERNEL]->select_read( $inotify_FH, "inotify_poll" );
   },
   inotify_poll => sub {
    $_[HEAP]{inotify}->poll;
   },
    watch_hdlr => \&watch_hdlr,
   },
 );

 POE::Kernel->run();
}

exit 0;
Share Button

ISUG Technical Journal Sep/Oct 2008 is out! Includes my “Perl in the Shell” article

The ISUG Technical Journal Sep/Oct 2008 is out.

This issue contains:

  • Sybase IQ and Sybase WorkSpace Part II By Mike Crocker
  • Perl in the shell By Jason Froebe
  • Taming the process nightmare: Consolidating a Horde of Process Diagrams with PowerDesigner By Mike Nicewarner
  • Information overload: Weapons-grade data analysis comes of age By Teresa Foster
  • Maximizing PowerBuilder and SQLAnywhere Part III SQLAnywhere, a PowerBuilder Developer’s perspective
    By Chris Pollach
  • select random (stuff) from SYBASE Command shell by proxy By Bill Grant
  • Mobile Enterprise Everywhere: Taking Sybase on the road By Adrian Bridgwater
  • Sybase TechWave 2008: Time for the technology tsunami

You need to be an ISUG member to receive the ISUG Technical Journal.  Join ISUG today!

International Sybase User Group

International Sybase User Group

Share Button

Example code using Perl’s Linux::Inotify2 module with POE

The examples in Linux::Inotify2 are for Event, Glib::IO and a manual loop. With the help of rcaputo, tye and Animator over on Perlmonks.org, I was able to get Linux::Inotify2 to work with POE. 🙂

#!/usr/bin/perl

use strict;
use warnings;

use Linux::Inotify2;
use POE;

$|++;

POE::Session->create
  ( inline_states =>
      { _start => sub {
            $_[KERNEL]->alias_set('notify');
            $_[HEAP]{inotify} = new Linux::Inotify2
              or die "Unable to create new inotify object: $!";

            $_[HEAP]{inotify}->watch("/tmp/j", IN_CLOSE_WRITE, $_[SESSION]->postback("watch_hdlr"))
              or die "Unable to watch dir: $!";

            my $inotify_FH;
            open $inotify_FH, "< &=" . $_&#91;HEAP&#93;{inotify}->fileno or die "Can't fdopen: $!\n";
            $_[KERNEL]->select_read( $inotify_FH, "inotify_poll" );
        },
        inotify_poll => sub {
          $_[HEAP]{inotify}->poll;
        },
        watch_hdlr => \&watch_hdlr,
      },
  );

sub watch_hdlr {
  my $event = $_[ARG1][0];

  my $name = $event->fullname;

  print "$name was accessed\n" if $event->IN_ACCESS;
  print "$name is no longer mounted\n" if $event->IN_UNMOUNT;
  print "$name is gone\n" if $event->IN_IGNORED;
  print "$name is new\n" if $event->IN_CLOSE_WRITE;
  print "events for $name have been lost\n" if $event->IN_Q_OVERFLOW;
}

POE::Kernel->run();

exit 0;

Create files for testing:

dd if=/dev/zero of=/tmp/j/p bs=1M count=2
2+0 records in
2+0 records out
2097152 bytes (2.1 MB) copied, 0.112971 s, 18.6 MB/s

Output

./test_poe_linux_inotify.pl
/tmp/j/p is new
/tmp/j/p is new
Share Button

Perl, POE and the pesky reaper

If you’re like me, you might be using POE (Perl Object Environment) for Perl in your applications. Using Wheel::Run is quite easy once you realize that it is just a big event loop. The documentation is quite sparse but unless you look carefully, you might overlook how to avoid the “Child process PID:26862 reaped” messages that occur when your application finishes.

Yup, it bit me too. The solution is to set up a child signal handler in the POE kernel and not in the global namespace as you would normally do.

Here we add the child signal handler, we pretty much ignore it in this code but we might flesh it out with some meaningful messages or something. Note that “$poe_kernel->sig_handled();” tells the POE kernel that this particular signal has been handled.

sub signal_child {
    $poe_kernel->sig_handled();
}

$poe_kernel->sig(CHLD => \&signal_child);

Notice that to handle the interrupt signal (ctrl-c), we need to use the global handler so that we can close down the application:

sub signal_int {
    print "\n\nERROR: Aborting on interrupt signal\n";
    exit -1;
}

$SIG{INT} = \&signal_int;

That’s it 🙂

Share Button

HOWTO: Running multiple processes in a single Perl script using POE

The perl module POE is best known for simulating multiple user threads in a perl application.

POE is also very useful for managing the forking and managing of child processes.  By using POE to handle this, the programmer can stop worrying about correctly handling the forking.

Set the number of child processes with MAX_CONCURRENT_TASKS.  For example, to allow for three child processes that do the work:

sub MAX_CONCURRENT_TASKS () { 3 }

The child process executes the do_stuff() function, so put whatever you need to run in parallel there.

Continue reading

Share Button