Hi, what I want to do is write some daemon with a defined count of worker processes. To keep things readable I will only post a minimal example of the problem I'm facing (some kind of worker queue):
!/usr/bin/perl use strict; use warnings; use Async; my $workercount = 10; my @workerqueue; for my $i (1..$workercount) { my $proc = Async->new( sub { sleep($i*3); print "$i\n"; } ); print "--> enqueue $proc->{PID}\n"; push @workerqueue, $proc; } while (1) { sleep(1); my @unfinished; while (my $proc = shift @workerqueue) { if ($proc and $proc->ready) { print "--> $proc->{PID} is ready\n"; if (my $e = $proc->error) { print "Something went wrong. The error was: $e\n"; } undef $proc; } else { push @unfinished, $proc; } } push @workerqueue, @unfinished; ## START refill queue block ## my $free = $workercount - @workerqueue; for my $i (1..$free) { my $proc = Async->new( sub { sleep($i*3); print "$i new\n"; } +); print "--> enqueue $proc->{PID}\n"; push @workerqueue, $proc; } ## END refill queue block ## print "mainloop\n"; }
Actually if I comment out the "refill queue block" everything works as expected, the output looks something like:
--> enqueue 28304 --> enqueue 28305 --> enqueue 28306 --> enqueue 28307 --> enqueue 28308 --> enqueue 28309 --> enqueue 28310 --> enqueue 28311 --> enqueue 28312 --> enqueue 28313 mainloop mainloop 1 --> 28304 is ready mainloop mainloop mainloop 2 --> 28305 is ready mainloop mainloop mainloop 3 --> 28306 is ready mainloop mainloop mainloop 4 --> 28307 is ready mainloop mainloop mainloop 5 --> 28308 is ready mainloop mainloop mainloop 6 --> 28309 is ready mainloop mainloop mainloop 7 --> 28310 is ready ...
But as soon as I include the refill queue block it looks like:
--> enqueue 28319 --> enqueue 28320 --> enqueue 28321 --> enqueue 28322 --> enqueue 28323 --> enqueue 28324 --> enqueue 28325 --> enqueue 28326 --> enqueue 28327 --> enqueue 28328 mainloop mainloop 1 --> 28319 is ready --> enqueue 28329 mainloop mainloop mainloop 2 1 new --> 28320 is ready --> 28327 is ready --> 28328 is ready --> 28329 is ready --> enqueue 28330 --> enqueue 28331 --> enqueue 28332 --> enqueue 28333 mainloop --> 28321 is ready --> 28322 is ready --> 28323 is ready --> 28324 is ready --> 28325 is ready --> 28326 is ready --> enqueue 28334 --> enqueue 28335 --> enqueue 28336 --> enqueue 28337 --> enqueue 28338 --> enqueue 28339 mainloop mainloop 1 new --> 28330 is ready --> enqueue 28340 ...
So what is happening here? As far as my debugging got me, I can tell that for any reason all running processes are considered to be ->ready() as soon as I start a new process out of the main loop. This leads to cleanup and therefore process being killed, so that it never gets a chance to print its output, after that a new process is created in the workerqueue. Since all previous processes are considered ready at a time this leads to the "--> enqueue ..." spam.
Now the really interesting question: What am I doing wrong so that Async module considers my processes been ready?
Note: I tried reading through the Async module and as for lets say 90% I get what it is doing. I tried commenting out the following line, but it didn't change anything:
kill 9 => $pid; # I don't care.
EDIT: Refering to http://www.perlmonks.org/?node_id=966939 , the example posted by Tanktalus sounds absolutely like what I need. However looking at http://perldoc.perl.org/perlthrtut.html and considering the hint of arpad.szasz: "It seems You are mixing old-style and deprecated threads model(Thread module) with the new ithreads thread model..." I'm wondering. Isn't Tanktalus' example therefore also using the depracated Version of Threads? Is there an equivalent implementaion with the newer ithreads? And could this code be modified to share (or better return to) data with its parent process?
EDIT 2: I think I will now use something like this (unless anyone has a good reason not to do so):
#!/usr/bin/perl use strict; use warnings; use threads; use threads::shared; use Data::Dumper; my %state :shared; my $workercount = 10; my $run = 1; $SIG{TERM} = sub { $run = 0; }; $SIG{INT} = sub { $run = 0; }; my @workerqueue; for my $i (1..$workercount) { my $thr = threads->create(\&mysub, $i); my $id = $thr->tid(); print "--> enqueue $id\n"; $state{$id} = 'running'; push @workerqueue, $thr; } while ($run) { sleep(1); my @unfinished; while (my $thr = shift @workerqueue) { my $id = $thr->tid(); if ($thr and $state{$id} eq 'finished') { print "--> $id is ready\n"; my $x = $thr->join(); delete $state{$id}; print Dumper $x; # do something useful with the data } else { push @unfinished, $thr; } } push @workerqueue, @unfinished; ## START refill queue block ## my $free = $workercount - @workerqueue; for my $i (1..$free) { my $thr = threads->create(\&mysub, $i); my $id = $thr->tid(); print "--> enqueue $id\n"; $state{$id} = 'running'; push @workerqueue, $thr; } ## END refill queue block ## print "mainloop\n"; } my @threads = threads->list(); foreach my $thr (@threads) { $thr->join(); } sub mysub { my $i = shift; $i *= 3; sleep $i; my $id = threads->tid(); print "$id : slept for $i sec\n"; my $x = { 'ID' => $id, 'a' => [1,2,3], 'B' => { 'a' => 'A', }, }; $state{$id} = 'finished'; return $x; }
In reply to Problems using module Async by hfi
| For: | Use: | ||
| & | & | ||
| < | < | ||
| > | > | ||
| [ | [ | ||
| ] | ] |