Re^6: threads on Windows

I've thought about what you've said and played with code for a few days. And I come up with some new material, posted below. I have multiple threads, which I call king, minister, worker, and media. The king (the main thread) makes proclamations, i.e. takes keyboard input and sticks it on the input queue. The minister implements these by sitting on the input queue, deciding policy and delegating to workers or sticking errors on the output queue. The worker threads are unchanged from the code you posted a few nodes up. All output is handled by the media thread, which sits on the output queue. In this way, I tried to keep the message agents (minister and media) working only on one queue. If this does gain a GUI, the king would be where this would be implemented.

There are a few problems I see with what I have written:

There's metadata problem. While the current system only has minimal information flow through the system, a more complex problem is likely to have a significant amount of information that won't matter to the intensive computation but must maintain its connection to maintain context. I can see two clear solutions - pass an object through the entire system (Minister->worker->Media) or create shared hashes between Minister and Media. I'm inclined toward the former from an aesthetic perspective.
There's a small risk in the status command of locking the minister thread if the target is paused. I tried to restructure things to minimize that risk, but it's still non-zero.
I tried to keep the command codes localized, but I had to modify the king loop condition from $minister->is_running to not /^q/ to avoid having to enter an extraneous <enter> to exit the script.

As a final note, if I do not explicitly close all thread-inappropriate channels before proceeding, I get blocking behavior between the <> in the King thread and the new thread generation in the Minister thread. I still do not grok why this happens - if you know of any good materials that discuss how I/O is implemented on a MS system, particularly with regard to Perl behaviors, I'd appreciate a shove in that direction. I hate solutions that say "Just don't do that".

Any wisdom you care to share would be appreciated.

#! perl -slw
use strict;
use threads;
use Thread::Queue;
use Thread::Semaphore;

my $Q_outbox = new Thread::Queue;
my $Q_inbox  = new Thread::Queue;
my $gag = new Thread::Semaphore(0);

my $media = new threads(\&media,
                        $Q_outbox,
                        $gag
                        );
close STDOUT;
close STDERR;
$Q_outbox->enqueue("media initiated...");

my $minister = new threads(\&minister,
                           $Q_inbox,
                           $Q_outbox,
                           $gag,
                           );
$Q_outbox->enqueue("minister initiated...");

# It's good to be king
$gag->up(2);

#while ($minister->is_running) {
$_ = 'something irrelevant';
while (not /^q/) {
    $_ = <STDIN>;
    chop;
    if (length) {
        $Q_inbox->enqueue( $_ );
    }
}

# Clean up
$minister->join;
$media->join;

#-----------------------------------------------------------
sub media {
# Condenses input from many sources into one stream
    my ($queue, $gag) = @_;
    close STDIN;
    $gag->down;

    while ( defined( $_ = $queue->dequeue ) ) {
        print STDOUT ">$_";
    }
}
#-----------------------------------------------------------
sub minister {
# Receives orders from king and either dispatches specific
# requests to workers or complains to media
    my ($Q_inbox, $Q_outbox, $gag) = @_;
    my %workers;
    my %sems;
    close STDIN;
    $gag->down;    

    while (my $order = $Q_inbox->dequeue) {
        local $_; my $name;
        if ($order =~ m[^([a-zA-Z])[a-zA-Z]*\s*(\d+)?]) {
            $_ = $1;
            $name = $2;
        } else {
            $order ||= ' ';
            $Q_outbox->enqueue("Bad command $order");
            next;
        }
        
        if (m[^q]i) {
            for( keys %workers ) {
                $sems{ $_ }->up;
                $workers{ $_ }->kill( 'KILL' );
            }
            $_->join for values %workers;
            $Q_outbox->enqueue( undef ); # Terminate main loop
            last;
        }

        if (not defined $name) {
            $Q_outbox->enqueue("Non-integer thread name in '$order'");
            next;
        }

        if( m[^w]i ) {
            if (not defined $name) {
                $Q_outbox->enqueue("You must give a name for the worke
+r");
                next;
            }
            if (exists $workers{ $name }) {
                $Q_outbox->enqueue("Worker $name already exists");
                next;
            }
            $sems{ $name } = new Thread::Semaphore;
            $workers{ $name } = threads->create( \&worker,
                                                $name, $Q_outbox, $sem
+s{ $name } 
                                                );
            next;
        }

        if (not exists $workers{ $name }) {
            $Q_outbox->enqueue("Worker $name does not exist");
            next;
        }

        if (m[^k]i) {
            $sems{ $name }->up;  ## Ensure the worker can respond
            delete( $workers{ $name } )->kill( 'KILL' )->join;
            next;
        }

        if( m[^p]i) {
            ${ $sems{ $name } }
                ? $sems{ $name }->down
                : $sems{ $name }->up;
            $workers{ $name }->kill( 'STOP' );
            next;
        }

        if (m[^s]i) {
            $workers{ $name }->kill( 'FPE' );
            $sems{ $name }->up; $sems{ $name }->down;
            next;
        }

        $Q_outbox->enqueue("Bad command $order");
    }
}
#-----------------------------------------------------------
sub worker {
# Follow orders and report results to the media
    my( $name, $queue, $sem ) = @_;
    my $cooldown = 0;
    my $state = 0;

    $SIG{KILL} = sub {
        $queue->enqueue("Worker $name dying");
        threads->exit();
    };
    $SIG{STOP} = sub {
        my $string = "Worker $name is ";
        $string   .= ( $state ^= 1 ) ? 'paused' : 'resumed';
        $queue->enqueue($string);
        $sem->down(); $sem->up();
    };
    $SIG{FPE } = sub {
        $sem->up();
        $queue->enqueue("Worker $name has cooldown $cooldown");
        $sem->down(); 
    };

    $queue->enqueue("Worker $name starting");
    my %nodes = ();
    while (1) {
        $queue->enqueue("$name step $cooldown");
        sleep 1;
        $cooldown++;
    }
}
[download]

Comment on Re^6: threads on Windows Select or Download Code

Replies are listed 'Best First'.
Re^7: threads on Windows by BrowserUk (Patriarch) on Feb 18, 2009 at 15:22 UTC
It's still not clear to me what benefit you think you are getting from passing the inputs to another thread to be actioned? I don't yet have a multi-core machine to test my hypothysis, but I strongly suspect that even if your minister thread is currently running on a another core when you read and post the input to the queue, that it probably takes an order of magnitude longer for the code to get actioned than if you did it inline within the read loop. If you are wedded to your "king's don't do things for themselves" analogy, then just call your minister subroutine directly, rather messing around with an extra thread and queue. I don't doubt it will work okay your way, I just don't see any benefit coming from the extra complexity and the performance hit it entails? I'm also unconvinced about the need for semaphores and signals in your workers. I see what your doing with them, and that it works, but still the question remains: why? Why do you need to pause/resume/kill threads? So far, the whole exercise seems to be a case of: "There are these modules kicking around, I wonder what I can do with them."; rather than: "I have this problem to solve, how can I best achieve that.". Ie. You seem to be trying to develop and infrastructure to cater to some supposed possible requirements, rather trying to code a solution to a specific problem. In every case where I've seen a framework developed in the absence of a real and well specified requirement, the framework has ended up being overengineered, inefficient and a nightmare to use for real applications, because it was built to satisfy some theoretical design goal rather than real application requirements. For example: Kings rarely instruct ministers; rather they simply endorse the proposals put forward by ministers. And kings certainly don't answer there own phones--or whatever anology you use for STDIN. And shouldn't you have a few secretaries around to do the donkey work? I really don't think that I can be of any further help to you until you describe the actual requirements of your real application, rather than some badly fitting analogy. Examine what is said, not who speaks -- Silence betokens consent -- Love the truth but pardon error. "Science is about questioning the status quo. Questioning authority". In the absence of evidence, opinion is indistinguishable from prejudice. "Too many [] have been sedated by an oppressive environment of political correctness and risk aversion."	[reply]
Re^8: threads on Windows by kennethk (Abbot) on Feb 18, 2009 at 16:38 UTC
If the following post is just more line noise, I won't be offended if you choose not to reply. The assumed benefit is creation of an abstract API. The performance hit taken with the extra thread should be relatively negligible because tasks should be few and relatively long running. The minister and workers can be packaged in a library module while the king and media would be interface specific. The program I'm hoping to get updated is proprietary (and written in VB...) so I can't really show you examples. The talk around the group is to not only have the desktop version but also a simplified subscription-based web version, thus my desire for abstraction. And in any case, much of this was a case of "I wonder what I can do with them" to gain familiarity with Perl's thread syntax and capabilities before I put in a formal proposal.	[reply]
Re^9: threads on Windows by BrowserUk (Patriarch) on Feb 19, 2009 at 04:34 UTC
The assumed benefit is creation of an abstract API. The problem is, whether you action inputs in the current thread, or in another thread has nothing to do with abstraction: it is an implementation detail. And abstraction might be something like a subroutine called: `enact()`. It might be used something like: `while( my $decree = <STDIN> ) { chomp $decree; enact( $decree ); }` [download] You can now implement `enact()` is any number of ways: as immediate code as an if/elsif/else cascade, or a dispatch table; or passing the responsibility off to a concurrent thread or process via a queue, pipe, socket or shared memory. Whatever. But as currently coded, there is no abstraction. (Other than the use of a queue and another thread, and they add complexity rather than hide it!) And you are gaining no benefit from the use of concurrency. It will take at least 2 context switches to go from reading an input to the point of having enacted it. And a context switch will take 2 or 3 orders of magnitude longer (x100 or x1000) than processing the input inline. All you've achieved by multithreading that data path, is to use more memory and make things slower. Now given that all of the inputs you've outlined can be processed far more quickly than a human being can type the next command, there is no benefit there either. You would only benefit if the inputs were being supplied by another program or computer faster than you could process them inline. And even then, unless the input stream was bursty, all you would do is fill up the queue with commands faster than you could take them off and process them. Voltaire (apparently) said: a book is not finished when you can add no more, but rather when you can remove no more. And nothing is a more laudable and beneficial goal in code development, than removing anything that serves no purpose. Examine what is said, not who speaks -- Silence betokens consent -- Love the truth but pardon error. "Science is about questioning the status quo. Questioning authority". In the absence of evidence, opinion is indistinguishable from prejudice. "Too many [] have been sedated by an oppressive environment of political correctness and risk aversion."	[reply] [d/l] [select]
Re^10: threads on Windows by kennethk (Abbot) on Feb 19, 2009 at 14:38 UTC