mce_mojolicious.pl

#!/usr/bin/env perl

# mce_mojolicious.pl
# based on https://gist.github.com/tardisx/9088819
# using MCE::Hobo and MCE::Shared

=pod

Using Mojolicious and L<MCE> for forked independant workers,
communicating via L<MCE::Queue>.

Run this with C<./mce_mojolicious.pl daemon>, then visit C<http://localhost:3000>.

There are 4 queues:

* One for requests for files that should be MD5'd
* One for directory requests (which feeds into the first)
* One for the results of each MD5
* One for log entries

We can add files, or directories via the web interface. Even adding massive
directories does not cause a slowdown, as the directory is processed in a
separate process, which feeds the MD5 queue, again processed in another process.

The homepage gives us a dynamic view of the queues and the results.

L<Mojo::IOLoop> timers are used to pull results and logs out of the queues
into our web server process, without blocking.

=cut

use strict;
use warnings;

BEGIN {
   $ENV{'MOJO_REACTOR'} = 'Mojo::Reactor::Poll' if ($^O eq 'MSWin32');
}

use Mojolicious::Lite;
use Mojo::IOLoop;

use MCE::Hobo;
use MCE::Shared;

use Time::HiRes qw/time/;
use File::Find  qw/find/;
use Digest::MD5;

my $Q_md5_req = MCE::Shared->queue();  # queue for md5 requests
my $Q_dir_req = MCE::Shared->queue();  # queue for md5 requests
my $Q_md5_res = MCE::Shared->queue();  # queue for md5 responses
my $Q_log     = MCE::Shared->queue();  # log messages

##################################################################
##
##  Dir and MD5 roles
##
##################################################################

sub dir_role {
  $Q_log->enqueue(scalar(localtime) . " starting dir worker, PID: $$");

  # when directories come into the queue, add all the files within to the
  # md5 file request queue
  while (defined (my $dir = $Q_dir_req->dequeue)) {
    my $now = time();
    $Q_log->enqueue(scalar(localtime) . " begin: scanning dir $dir");

    find(sub { $Q_md5_req->enqueue($File::Find::name)
      if -f $File::Find::name && -r $File::Find::name }, $dir);
  }
}

sub md5_role {
  $Q_log->enqueue(scalar(localtime) . " starting md5 worker, PID: $$");

  # pull file off this queue and calculate the md5
  while (defined (my $req = $Q_md5_req->dequeue)) {
    my $now = time();
    $Q_log->enqueue(scalar(localtime) . " begin: md5 on $req");

    # calculate the md5
    my $md5 = Digest::MD5->new();
    open my $fh, "<", $req || next;
    binmode $fh;
    $md5->addfile($fh);

    # store the results
    $Q_md5_res->enqueue({ $req => $md5->hexdigest });
    $Q_log->enqueue(scalar(localtime) . " end: md5 on $req took " . (time() - $now));
  }
}

# spawn MCE Hobo workers
MCE::Hobo->create('dir_role') for ( 1 .. 1 );
MCE::Hobo->create('md5_role') for ( 1 .. 4 );

##################################################################
##
##  Main Mojolicious WebApp
##
##################################################################

my $md5s = {};
my @audit_log = ();

# watch our results queue, bring results into process as they
# appear
my $md5_Q_watch = Mojo::IOLoop->recurring( 1.0 => sub {
   my $loop = shift;

   while (defined (my $result = $Q_md5_res->dequeue_nb)) {
      $md5s  = { %$md5s, %$result };
   }
});

# also watch the log queue
my $log_Q_watch = Mojo::IOLoop->recurring( 2.0 => sub {
  while (defined (my $log = $Q_log->dequeue_nb)) {
    push @audit_log, $log;
  }
});

# log to a local file
app->log(Mojo::Log->new(path => 'mce_mojo2.log'));

# route for adding files/dirs to the request queue
get '/queue' => sub {
  my $self = shift;
  my $path = $self->param('add');
  my $msg;

  if (-f $path && -r $path) {
    $Q_md5_req->enqueue($path);
    $msg = "added $path";
  }
  elsif (-d $path) {
    $Q_dir_req->enqueue($path);
    $msg = "added directory $path";
  }
  else {
    $msg = "No such file or directory $path";
  }

  $Q_log->enqueue(scalar(localtime) . " $msg");
  $self->flash(msg => $msg);
  $self->redirect_to('/');
};

# our main status page
get '/' => sub {
  my $self = shift;
  $Q_log->enqueue(scalar(localtime) . " user requested /");
  $self->stash(pending => $Q_md5_req->pending);
  $self->stash(results => $md5s);
  $self->stash(logs    => \@audit_log);
  $self->render('index');
};

# log our startup
$Q_log->enqueue(scalar(localtime) . " starting web server, PID: $$");
app->start;

__DATA__

@@ index.html.ep
% layout 'default';
% title 'Mojolicious+MCE MD5 Server Status Page';

<h1>Mojolicious and MCE example</h1>

% if (flash('msg')) {
<p><b><%= flash('msg') %></b></p>
% }

<p>Enter a filename or directory to calculate MD5sums for.</p>
%= form_for queue => begin
%= text_field 'add'
%= submit_button
% end

<p>Reload this page for status updates and results.</p>

<h4>Queued <%= $pending %> / <%= scalar keys %$results %> computed so far</h4>

<h4>Results:</h4>
<table>
  <tr><th>file</th><th>size</th><th>md5</th>
% foreach (sort keys %$results) {
<tr><td><%= $_ %></td><td><%= -s $_ %></td><td><%= $results->{$_} %></td></tr>
% }
</table>

<h4>Logs:</h4>

<pre>
<%= join("\n", reverse @$logs) %>
</pre>

@@ layouts/default.html.ep
<!DOCTYPE html>
<html>
  <head><title><%= title %></title></head>
  <body><%= content %></body>
</html>