Dear
BrowserUk
Following , the current running code.
Please note that
.
Since there is still some queries that returns empty values, I must check the result before sending to the db queue.
When querie returns empty rssi , I put in the queue back and it takes between 3-8 times to get a valid value.
I also introduce some delay, in order to avoid a loop in this process.
In comment at line 28 , Im asking if are they going to the walker instead ?
That's all
.
Leandro.
#!/usr/bin/perl -slw
use strict;
use threads;
use Thread::Queue;
our $T = 20; ## 10 walkers; adjust to suit.
my $cpe;
my $community;
my $snmp_rssi;
my $output;
my $error = 'none';
my $rssi;
my @result;
my $line;
sub listener {
my( $Qout ) = @_;
require IO::Socket::INET::Daemon; ## Requiring here means other
+threads don't carry the redundant weight
my $host = new IO::Socket::INET::Daemon(
host => '172.24.3.208', port => 7777, timeout => 20,
callback => { data =>
sub {
my ($io, $host) = @_;
chomp( my $line = <$io> );
return 0 unless $line;
$Qout->enqueue( $line ); ## send work to listener (
+is not sending to walker ?)
return !0;
}
},
);
$host->run;
return;
}
sub walker {
my( $Qin, $Qout ) = @_;
# while( $Qin->dequeue ) { ## receive work from listener
while (defined(my $item = $Qin->dequeue())) { ## receive work from
+ listener
#print "here \$\_ is : $item";
###
my ($type, $ip, $mac, $bsid, $datecode) = split(',', $item);
print "on walker we have $ip | $mac | $bsid\n";
$cpe=$ip;
$mac=~s/-//g;
$community='public';
$output=qx(snmpwalk -v2c -t1 -c $community $cpe $snmp_rssi
+ 2>&1);
if( $output eq "Timeout: No Response from $ip" ) {
$rssi=0;
$error='SNMP not responding. Upgrade firmware';
}
else
{
@result=split(/:/,$output);
$rssi=$result[3];
$rssi=~s/ //g;
$rssi=~s/\n//g;
if($rssi < -100) {
$rssi=$rssi/100;
}
$rssi=int($rssi);
}
if(($mac ne '') && ($rssi ne '')){ #### I must check rssi in
+order to avoid introducing empty values on db.
$Qout->enqueue(join(',',$mac,$ip,$bsid,$rssi));## data
+ items to DBI
}else{
$line = join(',',$type,$ip,$mac,$bsid);
print "will reenqueue $line\n";
sleep(5);
$Qin->enqueue($line);
}
}
}
use enum qw[ IN DBI_ENUM ];
my @Qs = map Thread::Queue->new(), 1 .. 2; # set up two Qs
## start the listener thread
my $tListener = threads->create( \&listener, $Qs[ IN ] ); ## One for t
+he listener to send work to the walkers
## start 10 walkers
my @walkers = map{ threads->create( \&walker, @Qs[ IN, DBI_ENUM ] ) }
+1 .. $T; ## And one for the walkers to forward data for adding to the
+ db
require DBI; ## Avoid loading DBI into threads
my $dbh = DBI->connect("DBI:mysql:database=cpe_info;host=172.24.3.207;
+port=3306","account_process","neting.!" );
my $sth = $dbh->prepare("INSERT INTO cpe_info(mac,ip,bsid,rssi) VALUE
+S
(?, ?, ?, ?) ON DUPLICATE KEY UPDATE ip = ?, bsid = ?, rssi =
+?");
## process data produced by walkers
#while( $Qs[DBI_ENUM]->dequeue ) {
while (defined(my $item = $Qs[DBI_ENUM]->dequeue())) { ## receive
+work from listener
my($mac, $ip, $bsid, $rssi) = split(',', $item); #Retrieve individ
+ual data
print "in db task item is $item";
$sth->execute($mac, $ip, $bsid, $rssi, $ip, $bsid, $rssi);
+ ## bind and execute
}
$dbh->disconnect();