Boss - Worker Threaded Model implemented in Perl
April 5, 2007 - 4:58am — Joshua Radke
I've completed the Boss - Worker model which I'll probably use for most of these problems that I'll run into. I couldn't find anything quite like this with Google, so I'm publishing it here, hopefully giving somebody else a head start next time they need to make their serial program multi-threaded. It's fairly minimalistic ... no bells or whistles, just what's needed to do the threading.
#!/usr/bin/perl -w
#
# threadbossworker.pl
#
# Copyright 2007 Joshua Radke (josh at radkeland dot org)
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License (version 2) as
# published by the Free Software Foundation.
#
# This program is a demonstration (simple scaffold) for a boss/worker
# threaded program. It uses semaphores to control the number of worker
# threads allowed. It uses a general cleanthreads() routine to gather
# the results from finished threads. The cleanthreads routine takes a
# reference to a function to determine what to do with the harvested
# threads.
#
# These are the minimum needed to implement a semaphore managed
# threading program.
use strict;
use threads;
use Thread::Semaphore;
# Cleanthreads harvests and dispatches the return values of the worker
# threads (detail with the actual function). Idleworker hangs out
# awhile, and returns its original value untouched. Note that
# idleworker needs to be thread aware (increment and decrement the
# semaphore). Printreturn prints the value of whatever is passed to
# it, with a newline appended.
sub cleanthreads($$@);
sub idleworker($);
sub printreturn($);
my $maxthreads = 5;
my $sem = new Thread::Semaphore($maxthreads);
for my $i (1 .. 10) {
threads->create(\&idleworker, $i) or die "Unable to create thread";
cleanthreads(0, \&printreturn, "foo");
}
cleanthreads(1, \&printreturn, "foo");
exit(0);
# Cleanthreads is the interesting workhorse of the threading model. It is
# called like this:
# cleanthreads(<mode>, <action function>, <list of other arguments
# for action function> )
# Mode = 0 or 1. The meanings are:
# 0: Find all joinable threads , and join/process the return value.
# Do not return unless there is at least one semaphore available.
# 1: Clean all threads but the parent before returning.
sub cleanthreads($$@) {
my $mode = shift @_;
my $func = shift @_;
my @othervals = (defined($_[0])) ? (@_) : (undef);
my $thrcount;
my @thrlist;
# First, get our list of threads to join.
if ($mode == 0) {
# We will guarantee that at least one thread is joinable or we are
# not at maximum capacity by decrementing the semaphore (then
# incrementing it).
$sem->down;
$sem->up;
@thrlist = threads->list(threads::joinable);
} elsif ($mode == 1) {
@thrlist = threads->list(threads::all);
} else {
die "Value $mode for \$mode not allowed in cleanthreads";
}
# And do the deed - oddly, the parent doesn't seem to get included in
# @thrlist when generated as above. Keep an eye on this.
foreach my $thr (@thrlist) {
&$func($thr->join(), @othervals);
}
return;
}
sub idleworker($) {
$sem->down;
my $retval = shift @_;
sleep 2;
$sem->up;
return $retval;
}
<br>sub printreturn($) {
print "$_[0]\n";
return;
}