FHEM/fhem-mirror
2
0
Fork 0
mirror of https://github.com/fhem/fhem-mirror.git synced 2025-04-22 08:11:44 +00:00
Code Issues Releases Wiki Activity

PRESENCE/collectord: fix missing state information, when executing "now" command and one room is not connected

Browse Source 
git-svn-id: https://svn.fhem.de/fhem/trunk@11211 2b470e98-0d58-463d-a4d8-8e2adae1ed80
This commit is contained in:
markusbloch 2016-04-09 16:10:27 +00:00
parent ab60b4b2ee
commit 6c353f15a2
1 changed files with 249 additions and 233 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

406
fhem/contrib/PRESENCE/collectord
View File

@ -70,6 +70,22 @@ my %socket_to_handle;
my $uuid; my $uuid;
$SIG{__DIE__} = sub {
my ($msg) = @_;
Log 1, "PERL ERROR: $msg";
};
$SIG{__WARN__} = sub {
my ($msg) = @_;
Log 1, "PERL WARN: $msg";
};
Getopt::Long::Configure('bundling'); Getopt::Long::Configure('bundling');
GetOptions( GetOptions(
"d" => \$opt_d, "daemon" => \$opt_d, "d" => \$opt_d, "daemon" => \$opt_d,
@ -107,15 +123,15 @@ sub print_usage () {
if($opt_h) if($opt_h)
{ {
print_usage(); print_usage();
exit; exit;
} }
if(-e "$opt_P") if(-e "$opt_P")
{ {
print STDERR timestamp()." another process already running (PID file found at $opt_P)\n"; print STDERR timestamp()." another process already running (PID file found at $opt_P)\n";
print STDERR timestamp()." aborted...\n"; print STDERR timestamp()." aborted...\n";
exit 1; exit 1;
} }
if(not $opt_c) if(not $opt_c)
@ -141,7 +157,7 @@ readConfig($opt_c);
if($opt_d) if($opt_d)
{ {
daemonize(); daemonize();
} }
# Write PID file # Write PID file
@ -204,27 +220,27 @@ Log 2, "finished initialization. entering main loop";
while(1) while(1)
{ {
# Cleaning up the status hash for obsolete devices # Cleaning up the status hash for obsolete devices
foreach $uuid (keys %state) foreach $uuid (keys %state)
{ {
my %handle_to_socket = reverse %socket_to_handle; my %handle_to_socket = reverse %socket_to_handle;
unless(exists($handle_to_socket{$uuid})) unless(exists($handle_to_socket{$uuid}))
{ {
Log 2, "cleaning up status values (UUID: $uuid)"; Log 2, "cleaning up status values (UUID: $uuid)";
delete $state{$uuid}; delete $state{$uuid};
} }
} }
# process all status messages from all threads via status queue # process all status messages from all threads via status queue
while($status_queue->pending) while($status_queue->pending)
{ {
($uuid,$room,$value,$name) = split(";", $status_queue->dequeue); ($uuid,$room,$value,$name) = split(";", $status_queue->dequeue);
Log 2, "processing state message for device ".(defined($name)?$name." ":"")."in room $room (UUID: $uuid)"; Log 2, "processing state message for device ".(defined($name)?$name." ":"")."in room $room (UUID: $uuid)";
if(not $value =~ /^(absence|present)$/) if(not $value =~ /^(absence|present)$/)
{ {
$handle{$uuid}{client}->send("$value;$room\n") if(defined($handle{$uuid}{client})); $handle{$uuid}{client}->send("$value;$room\n") if(defined($handle{$uuid}{client}));
if($value eq "socket_closed") if($value eq "socket_closed")
@ -251,99 +267,99 @@ while(1)
} }
} }
} }
#print Dumper(%state); #print Dumper(\%state);
} }
# If a thread has something reported via Log Queue, print it out if verbose is activated # If a thread has something reported via Log Queue, print it out if verbose is activated
while($log_queue->pending) while($log_queue->pending)
{ {
$logline = $log_queue->dequeue; $logline = $log_queue->dequeue;
Log 2, $logline; Log 2, $logline;
$logline = undef; $logline = undef;
} }
# If a INET socket has anything to report # If a INET socket has anything to report
if(@new_handles = $listener->can_read(1)) if(@new_handles = $listener->can_read(1))
{ {
foreach my $client (@new_handles) foreach my $client (@new_handles)
{ {
# if the socket is the server socket, accept new client and add it to the socket selector # if the socket is the server socket, accept new client and add it to the socket selector
if($client == $server) if($client == $server)
{ {
$new_client = $server->accept(); $new_client = $server->accept();
$listener->add($new_client); $listener->add($new_client);
Log 1, "new connection from ".$new_client->peerhost().":".$new_client->peerport(); Log 1, "new connection from ".$new_client->peerhost().":".$new_client->peerport();
} }
else # else is must be a client, so read the message and process it else # else is must be a client, so read the message and process it
{ {
$buf = ''; $buf = '';
$buf = <$client>; $buf = <$client>;
# if the message is defined, it is a real message, else the connection is closed (EOF) # if the message is defined, it is a real message, else the connection is closed (EOF)
if($buf) if($buf)
{ {
# replace leading and trailing white spaces # replace leading and trailing white spaces
$buf =~ s/(^\s*|\s*$)//g; $buf =~ s/(^\s*|\s*$)//g;
# if the message is a new command, accept the command and create threads for all rooms to process the command # if the message is a new command, accept the command and create threads for all rooms to process the command
if($buf =~ /^\s*([0-9a-fA-F]{2}:){5}[0-9a-fA-F]{2}\s*\|\s*\d+\s*$/) if($buf =~ /^\s*([0-9a-fA-F]{2}:){5}[0-9a-fA-F]{2}\s*\|\s*\d+\s*$/)
{ {
# send the acknowledgment back to the sender # send the acknowledgment back to the sender
$client->send("command accepted\n"); $client->send("command accepted\n");
Log 2, "received new command from ".$client->peerhost().":".$client->peerport()." - $buf"; Log 2, "received new command from ".$client->peerhost().":".$client->peerport()." - $buf";
# Split the message into bluetooth address and the timeout value # Split the message into bluetooth address and the timeout value
# (timeout is ignored within the collectord, as it is given by configuration) # (timeout is ignored within the collectord, as it is given by configuration)
($address, $timeout) = split("\\|", $buf); ($address, $timeout) = split("\\|", $buf);
# remove any containing white spaces # remove any containing white spaces
$address =~ s/\s*//g; $address =~ s/\s*//g;
$timeout =~ s/\s*//g; $timeout =~ s/\s*//g;
# if the client has already a request running, stop at first the old request # if the client has already a request running, stop at first the old request
if(defined($socket_to_handle{$client})) if(defined($socket_to_handle{$client}))
{ {
$uuid = $socket_to_handle{$client}; $uuid = $socket_to_handle{$client};
# get all threads for this socket and send them a termination signal # get all threads for this socket and send them a termination signal
my $temp = $handle{$uuid}{threads}; my $temp = $handle{$uuid}{threads};
foreach $room (keys %$temp) foreach $room (keys %$temp)
{ {
Log 2, "sending thread ".$handle{$uuid}{threads}{$room}->tid()." new address $address for room $room"; Log 2, "sending thread ".$handle{$uuid}{threads}{$room}->tid()." new address $address for room $room";
$queues{$handle{$uuid}{threads}{$room}->tid()}->enqueue("new|$address"); $queues{$handle{$uuid}{threads}{$room}->tid()}->enqueue("new|$address");
$state{$uuid}{rooms}{$room} = "" $state{$uuid}{rooms}{$room} = ""
} }
$handle{$uuid}{timeout} = $timeout; $handle{$uuid}{timeout} = $timeout;
$state{$uuid}{lastresult}{timestamp} = 0; $state{$uuid}{lastresult}{timestamp} = 0;
} }
else else
{ {
# create a new uuid if not exist for socket # create a new uuid if not exist for socket
if(not defined($socket_to_handle{$client})) if(not defined($socket_to_handle{$client}))
{ {
$socket_to_handle{$client} = generateUUID(); $socket_to_handle{$client} = generateUUID();
Log 2, "generating new UUID for client ".$client->peerhost()." - ".$socket_to_handle{$client}; Log 2, "generating new UUID for client ".$client->peerhost()." - ".$socket_to_handle{$client};
} }
$uuid = $socket_to_handle{$client}; $uuid = $socket_to_handle{$client};
$handle{$uuid}{address} = $address; $handle{$uuid}{address} = $address;
$handle{$uuid}{client} = $client; $handle{$uuid}{client} = $client;
$handle{$uuid}{timeout} = $timeout; $handle{$uuid}{timeout} = $timeout;
$state{$uuid}{lastresult}{value} = "absence"; $state{$uuid}{lastresult}{value} = "absence";
$state{$uuid}{lastresult}{timestamp} = 0; $state{$uuid}{lastresult}{timestamp} = 0;
# create a new reqester thread for each configured room to perform the query # create a new reqester thread for each configured room to perform the query
while (($room, $value) = each %config) while (($room, $value) = each %config)
{ {
$thread_counter++; $thread_counter++;
$queues{$thread_counter} = Thread::Queue->new(); $queues{$thread_counter} = Thread::Queue->new();
@ -356,115 +372,115 @@ while(1)
# save the socket/room relationship to know which thread belongs to which client request (for stop command) # save the socket/room relationship to know which thread belongs to which client request (for stop command)
$handle{$uuid}{threads}{$room} = $new_thread; $handle{$uuid}{threads}{$room} = $new_thread;
$state{$uuid}{rooms}{$room} = ""; $state{$uuid}{rooms}{$room} = "";
} }
} }
} }
elsif(lc($buf) =~ /^\s*now\s*$/) # if a now command is received, all threads need to be signaled to send a now command to the presenced server elsif(lc($buf) =~ /^\s*now\s*$/) # if a now command is received, all threads need to be signaled to send a now command to the presenced server
{ {
Log 2, "received now command from client ".$client->peerhost(); Log 2, "received now command from client ".$client->peerhost();
# just to be sure if the client has really a running request # just to be sure if the client has really a running request
if(defined($socket_to_handle{$client})) if(defined($socket_to_handle{$client}))
{ {
$uuid = $socket_to_handle{$client}; $uuid = $socket_to_handle{$client};
# get all threads for this socket and send them a termination signal # get all threads for this socket and send them a termination signal
my $temp = $handle{$uuid}{threads}; my $temp = $handle{$uuid}{threads};
foreach $room (keys %$temp) foreach $room (keys %$temp)
{ {
Log 2, "signalling thread ".$handle{$uuid}{threads}{$room}->tid()." to send \"now\"-request for room $room for client ".$client->peerhost(); Log 2, "signalling thread ".$handle{$uuid}{threads}{$room}->tid()." to send \"now\"-request for room $room for client ".$client->peerhost();
$queues{$handle{$uuid}{threads}{$room}->tid()}->enqueue("now"); $queues{$handle{$uuid}{threads}{$room}->tid()}->enqueue("now");
$state{$uuid}{rooms}{$room} = ""; $state{$uuid}{rooms}{$room} = "" if(exists($state{$uuid}{rooms}{$room}));
} }
$state{$uuid}{lastresult}{timestamp} = 0; delete($state{$uuid}{lastresult}) if(exists($state{$uuid}{lastresult}));
$client->send("command accepted\n"); $client->send("command accepted\n");
} }
else else
{ {
# if there is no command running, just tell the client he's wrong # if there is no command running, just tell the client he's wrong
$client->send("no command running\n"); $client->send("no command running\n");
} }
} }
elsif(lc($buf) =~ /^\s*stop\s*$/) # if a stop command is received, the running request threads must be stopped elsif(lc($buf) =~ /^\s*stop\s*$/) # if a stop command is received, the running request threads must be stopped
{ {
Log 1, "received stop command from client ".$client->peerhost(); Log 1, "received stop command from client ".$client->peerhost();
# just to be sure if the client has really a running request # just to be sure if the client has really a running request
if(defined($socket_to_handle{$client})) if(defined($socket_to_handle{$client}))
{ {
$uuid = $socket_to_handle{$client}; $uuid = $socket_to_handle{$client};
# get all threads for this socket and send them a termination signal # get all threads for this socket and send them a termination signal
my $temp = $handle{$uuid}{threads}; my $temp = $handle{$uuid}{threads};
foreach $room (keys %$temp) foreach $room (keys %$temp)
{ {
Log 2, "killing thread ".$handle{$uuid}{threads}{$room}->tid()." for room $room for client ".$client->peerhost(); Log 2, "killing thread ".$handle{$uuid}{threads}{$room}->tid()." for room $room for client ".$client->peerhost();
$queues{$handle{$uuid}{threads}{$room}->tid()}->enqueue("stop"); $queues{$handle{$uuid}{threads}{$room}->tid()}->enqueue("stop");
delete($handle{$uuid}{threads}{$room}); delete($handle{$uuid}{threads}{$room});
} }
# when all threads are signaled, delete all relationship entry for this client # when all threads are signaled, delete all relationship entry for this client
delete($handle{$uuid}); delete($handle{$uuid});
delete($socket_to_handle{$client}); delete($socket_to_handle{$client});
$client->send("command accepted\n"); $client->send("command accepted\n");
} }
else else
{ {
# if there is no command running, just tell the client he's wrong # if there is no command running, just tell the client he's wrong
$client->send("no command running\n"); $client->send("no command running\n");
} }
} }
else else
{ # if the message does not match a regular command or a stop signal, just tell the client and make a entry for logging. { # if the message does not match a regular command or a stop signal, just tell the client and make a entry for logging.
$client->send("command rejected\n"); $client->send("command rejected\n");
Log 1, "received invalid command >>$buf<< from client ".$client->peerhost(); Log 1, "received invalid command >>$buf<< from client ".$client->peerhost();
} }
} }
else # if the message is not defined (EOF) the connection was closed. Now let's clean up else # if the message is not defined (EOF) the connection was closed. Now let's clean up
{ {
# make a log entry and remove the socket from the socket selector # make a log entry and remove the socket from the socket selector
Log 1, "closed connection from ".$client->peerhost(); Log 1, "closed connection from ".$client->peerhost();
$listener->remove($client); $listener->remove($client);
# if there is a running command, stop it first and clean up (same as stop command, see above) # if there is a running command, stop it first and clean up (same as stop command, see above)
if(defined($socket_to_handle{$client})) if(defined($socket_to_handle{$client}))
{ {
$uuid = $socket_to_handle{$client}; $uuid = $socket_to_handle{$client};
# get all threads for this socket and send them a termination signal # get all threads for this socket and send them a termination signal
my $temp = $handle{$uuid}{threads}; my $temp = $handle{$uuid}{threads};
foreach $room (keys %$temp) foreach $room (keys %$temp)
{ {
Log 2, "killing thread ".$handle{$uuid}{threads}{$room}->tid()." for room $room for client ".$client->peerhost(); Log 2, "killing thread ".$handle{$uuid}{threads}{$room}->tid()." for room $room for client ".$client->peerhost();
$queues{$handle{$uuid}{threads}{$room}->tid()}->enqueue("stop"); $queues{$handle{$uuid}{threads}{$room}->tid()}->enqueue("stop");
delete($handle{$uuid}{threads}{$room}); delete($handle{$uuid}{threads}{$room});
} }
# when all threads are signaled, delete all relationship entry for this client # when all threads are signaled, delete all relationship entry for this client
delete($handle{$uuid}); delete($handle{$uuid});
delete($socket_to_handle{$client}); delete($socket_to_handle{$client});
} }
# now close the socket, that's it # now close the socket, that's it
close $client; close $client;
} }
} }
} }
} }
# in case we have received a process signal, remove the pid file and shutdown # in case we have received a process signal, remove the pid file and shutdown
if(defined($sig_received)) if(defined($sig_received))
{ {
Log 1, "Caught $sig_received exiting"; Log 1, "Caught $sig_received exiting";
unlink($opt_P); unlink($opt_P);
Log 1, "removed PID-File $opt_P"; Log 1, "removed PID-File $opt_P";
Log 1, "server shutdown"; Log 1, "server shutdown";
exit; exit;
} }
} }
Log 2, "leaving main loop"; Log 2, "leaving main loop";
@ -490,7 +506,7 @@ sub daemonize
} }
elsif($pid) elsif($pid)
{ {
Log 0, "forked with PID $pid"; Log 0, "forked with PID $pid";
exit 0; exit 0;
} }
@ -532,7 +548,7 @@ sub doQuery($$$)
Type => SOCK_STREAM, Type => SOCK_STREAM,
KeepAlive => 1, KeepAlive => 1,
Blocking => 1 Blocking => 1
) or ( $log_queue->enqueue(threads->tid()."|$room : could not create socket to ".$values{address}." - $! -")); ) or ( $log_queue->enqueue(threads->tid()."|$room : could not create socket to ".$values{address}." - $! -"));
$selector = IO::Select->new($client_socket); $selector = IO::Select->new($client_socket);
@ -556,7 +572,7 @@ sub doQuery($$$)
$log_queue->enqueue(threads->tid()."|$room socket to ".$values{address}.":".$values{port}." did not report anything in expected time, resetting socket (last contact: ".strftime("%Y-%m-%d %H:%M:%S", localtime($last_contact)).")"); $log_queue->enqueue(threads->tid()."|$room socket to ".$values{address}.":".$values{port}." did not report anything in expected time, resetting socket (last contact: ".strftime("%Y-%m-%d %H:%M:%S", localtime($last_contact)).")");
$selector->remove($client_socket); $selector->remove($client_socket);
shutdown($client_socket, 2); $client_socket->shutdown(2);
close($client_socket); close($client_socket);
$client_socket = undef; $client_socket = undef;
} }
@ -574,7 +590,7 @@ sub doQuery($$$)
elsif($cmd eq "stop") elsif($cmd eq "stop")
{ {
$log_queue->enqueue(threads->tid()."|$room terminating thread ".threads->tid()." for $address"); $log_queue->enqueue(threads->tid()."|$room terminating thread ".threads->tid()." for $address");
$client_socket->shutdown() if(defined($client_socket)); $client_socket->shutdown(2) if(defined($client_socket));
$selector->remove($client_socket) if(defined($selector)); $selector->remove($client_socket) if(defined($selector));
close($client_socket) if(defined($client_socket)); close($client_socket) if(defined($client_socket));
$client_socket = undef; $client_socket = undef;
@ -709,7 +725,7 @@ sub doQuery($$$)
$selector->remove($local_client); $selector->remove($local_client);
shutdown($local_client, 2); $local_client->shutdown(2);
close($local_client); close($local_client);
$client_socket = undef; $client_socket = undef;
} }