#!/usr/bin/perl # created by joe.zheng at Wed Apr 1 10:23:14 2009 use 5.008; use threads; # multi-threads needed use threads::shared; # need to share some variables use Thread::Queue; # thread safe communication needed use strict; use Getopt::Long; use HTTP::Daemon; use Time::HiRes qw( time ); # If available.... # sub proto type declaration sub trc(@); # print log message sub notify(@); # parameter: class, payload ... sub init_srv(); sub spawn_miner(); sub spawn_worker(); # global variables my $queue_up = Thread::Queue->new; my %stat : shared; my %data : shared; # configuration my %cnf = ( # Username/Password. usr => 'usr', pwd => 'pwd', # Appr. Position, being used for Ephemeris transfer from server. lat => 47.28, # Degrees lon => 8.56, # Degrees acc => 1000, # meters srv => 'agps.u-blox.com:46434', cmd => 'eph', # our server settings port => 46434, # listen port queue => 5, # listen queue timeout => 10, # rest => 10, # the time value to rest after previous conn +ect from the same addr conn => 3, # the allowed max connection per addr max_con => 20000, # max connection simultaneously ext_srv => '', # extent server to handle requests when t +he main server reach its capacity workers_ini => 8, # prethread count workers_lo => 2, # low water mark workers_hi => 10, # high water mark period => 600, # period to check for new ephemeris # directory doc => 'doc', # the root directory where all the docs live in # misc dbg => 1, # print log for debug purpose ); # parse command line parameter GetOptions(\%cnf, 'usr=s', 'pwd=s', 'lat=f', 'lon=f', 'acc=i', 'srv=s', 'cmd=s', 'period=f', 'port=i', 'queue=i', 'timeout=f', 'rest=f', 'conn=i', 'max_con=i', 'ext_srv=s', 'doc=s', 'dbg=i', ); # fix me, not work in win32, why? #$SIG{'INT'} = $SIG{'TERM'} = sub { lock %stat; $stat{'done'}++ }; my $listener = init_srv(); spawn_worker() foreach ( 1 .. $cnf{'workers_ini'} ); spawn_miner(); trc 'dbg', 'start main loop'; while ( !$stat{'done'} ) { if ( my $msg = $queue_up->dequeue() ) { trc 'dbg', 'new msg'; my ( $class, $payload ) = split /\s+/, $msg, 2; if ( $class eq 'work' ) { my ( $t, $s ) = split /\s+/, $payload, 2; lock %{ $stat{'work'} }; trc 'dbg', 'work stat', $t, $s; if ( $s eq 'gone' ) { delete $stat{'work'}{$t}; } else { $stat{'work'}{$t}{'stat'} = $s; $stat{'work'}{$t}{'time'} = time(); $stat{'work'}{$t}{'work'}++ if $s eq 'idle'; } } elsif ( $class eq 'data' ) { my ( $t, $d ) = split /\s+/, $payload, 2; if ( $d ) { lock %{ $stat{'data'} }; trc 'dbg', 'data', $t, length $d; $stat{'data'}{$t}{'expi'} = time() + $cnf{'period'}; $stat{'data'}{$t}{'data'} = $d; } } else { # nothing to do } } # reap worker if necessary # adjust worker my @worker_idle = sort { $a <=> $b } grep { $stat{'work'}{$_}{'sta +t'} eq 'idle' } keys %{$stat{'work'}}; trc 'dbg', 'idle worker', scalar @worker_idle; if ( @worker_idle < $cnf{'workers_lo'} ) { spawn_worker() for ( 1 .. $cnf{'workers_lo'} - @worker_idle ); } elsif ( @worker_idle > $cnf{'workers_hi'} ) { foreach my $t ( @worker_idle[ 0 .. @worker_idle - $cnf{'worker +s_hi'} - 1 ] ) { lock %{ $stat{'work'} }; $stat{'work'}{$t}{'stat'} = 'done'; } } } sub END { # close all the worker foreach ( keys %{$stat{'work'}} ) { lock %{ $stat{'work'} }; $stat{'work'}{$_}{'stat'} = 'done'; } } sub retrieve_data { # dummy data here return 'dummy data, comming soon'; } sub trc (@) { return if !$cnf{'dbg'} && $_[0] eq 'dbg'; local $, = ', '; print scalar(localtime), @_; print "\n"; } sub init_srv () { # print out the configuration for debug purpose if ( $cnf{'dbg'} ) { print '<configuration>', "\n"; foreach my $k ( sort keys %cnf ) { print $k, ': ', $cnf{ $k }, "\n"; } print '</configuration>', "\n"; } # main thread my $srv = HTTP::Daemon->new( LocalPort => $cnf{'port'}, Reuse => 1, Listen => $cnf{'queue'}, Timeout => $cnf{'timeout'}, ) or die "can't create local socket: $@\n"; trc 'log', "Accepting connections on", $srv->sockhost().':'.$srv-> +sockport(); { lock %stat; $stat{'lock'} = &share( {} ); $stat{'data'} = &share( {} ); $stat{'work'} = &share( {} ); } return $srv; } sub spawn_miner() { sub do_miner { # wait for the main thread finish initialization # fix me, yield can't achieve our purpose, why? # threads->yield(); # periodically check for ephemeris while ( !$stat{'done'} ) { # Check for new ephemeris notify( 'data', 'eph', retrieve_data() ); select undef, undef, undef, $cnf{'period'}; } } if ( my $thread = threads->new( \&do_miner ) ) { lock( %{ $stat{'data'} } ); $stat{'data'}{'eph'} = &share( {} ); $thread->detach; } } sub spawn_worker() { sub do_worker { my $t = threads->self()->tid(); # wait for the main thread finish initialization # fix me, yield can't achieve our purpose, why? # threads->yield(); sleep(1); my $c; LOOP: while ( !$stat{'work'}{$t}{'stat'} ne 'done' ) { { lock %{ $cnf{'lock'} }; $c = $listener->accept() or next LOOP; } notify( 'work', $t, 'busy' ); trc 'log', "new connect accept", $c->peerhost().":".$c->pe +erport(); # fix me, the thread blocks here, why? # if I lock %{ $cnf{'lock'} } untill close the $c, things +will be OK, # but then we can't gain advantage of the multi-threads no +r the pre-threads # design. if ( my $r = $c->get_request() ) { trc 'dbg', 'new request', $r->as_string; if ( $r->method() eq 'GET' ) { if ( $r->uri()->path() eq '/' ) { my %query = $r->uri()->query_form(); if ( exists $query{'cmd'} ) { my $data; for ( $query{'cmd'} ) { /eph/i and do { $data = $stat{'data'}{ +'eph'}{'data'}; last }; } if ( $data ) { my $res = HTTP::Response->new( 200 ); $res->push_header( 'Expires' => HTTP:: +Date->time2str( $stat{'data'}{'eph'}{'expi'} ) ); $res->push_header( 'Content-Type' => ' +application/ubx' ); $res->content( $data ); $c->send_response( $res ); trc 'log', 'response', 'ok', "send @{ + [ int( length $data ) ] } bytes"; } else { $c->send_error(); trc 'log', 'response', 'error'; } } else { foreach my $f ( glob "$cnf{'doc'}/index.*" + ) { trc 'dbg', 'send', $f; $c->send_file_response( $f ); last; } } } else { ( my $path = $r->uri()->path() ) =~ s|^/|$cnf{ +'doc'}/|; trc 'dbg', 'send', $path; $c->send_file_response( $path ); } } } $c->close(); notify( 'work', $t, 'idle' ); } notify( 'work', $t, 'gone' ); } if ( my $thread = threads->new( \&do_worker ) ) { my $t = $thread->tid; lock %{ $stat{'work'} }; $stat{'work'}{ $t } = &share( {} ); $stat{'work'}{ $t }{'stat'} = 'idle'; trc 'dbg', 'new worker created'; $thread->detach; } else { trc 'dbg', 'new worker created fail'; } } sub notify(@) { $queue_up->enqueue( join ' ', @_ ); }
In reply to multi thread problem in an simple HTTP server by bravesoul
| For: | Use: | ||
| & | & | ||
| < | < | ||
| > | > | ||
| [ | [ | ||
| ] | ] |