Collectd.pm: Use threads::shared to share @plugins between threads.

The list of plugins has to be common to all threads to be able to (un)register
callbacks after initialization. As threads::shared is not able to share
references to subroutines the callbacks now have to be identified by their
name. Collectd::call_by_name() is used to execute the callbacks.

Signed-off-by: Sebastian Harl <sh@tokkee.org>
Signed-off-by: Florian Forster <octo@huhu.verplant.org>

diff --git a/bindings/perl/Collectd.pm b/bindings/perl/Collectd.pm
index 1f9c61e..8c3159c 100644 (file)
--- a/bindings/perl/Collectd.pm
+++ b/bindings/perl/Collectd.pm
@@ -24,6 +24,9 @@ use warnings;
 
 use Config;
 
+use threads;
+use threads::shared;
+
 BEGIN {
        if (! $Config{'useithreads'}) {
                die "Perl does not support ithreads!";
@@ -75,7 +78,7 @@ our %EXPORT_TAGS = (
 
 Exporter::export_ok_tags ('all');
 
-my @plugins  = ();
+my @plugins : shared = ();
 
 my %types = (
        TYPE_INIT,     "init",
@@ -86,7 +89,7 @@ my %types = (
 );
 
 foreach my $type (keys %types) {
-       $plugins[$type] = {};
+       $plugins[$type] = &share ({});
 }
 
 sub _log {
@@ -109,6 +112,8 @@ sub DEBUG   { _log (scalar caller, LOG_DEBUG,   shift); }
 sub plugin_call_all {
        my $type = shift;
 
+       our $cb_name = undef;
+
        if (! defined $type) {
                return;
        }
@@ -122,9 +127,12 @@ sub plugin_call_all {
                return;
        }
 
+       lock @plugins;
        foreach my $plugin (keys %{$plugins[$type]}) {
                my $p = $plugins[$type]->{$plugin};
 
+               my $status = 0;
+
                if ($p->{'wait_left'} > 0) {
                        # TODO: use interval_g
                        $p->{'wait_left'} -= 10;
@@ -132,7 +140,18 @@ sub plugin_call_all {
 
                next if ($p->{'wait_left'} > 0);
 
-               if (my $status = $p->{'code'}->(@_)) {
+               $cb_name = $p->{'cb_name'};
+               $status = call_by_name (@_);
+
+               if (! defined $status) {
+                       if (TYPE_LOG != $type) {
+                               ERROR ("Could not execute callback \"$cb_name\": $@");
+                       }
+
+                       next;
+               }
+
+               if ($status) {
                        $p->{'wait_left'} = 0;
                        $p->{'wait_time'} = 10;
                }
@@ -194,13 +213,24 @@ sub plugin_register {
        if ((TYPE_DATASET == $type) && ("ARRAY" eq ref $data)) {
                return plugin_register_data_set ($name, $data);
        }
-       elsif ("CODE" eq ref $data) {
+       elsif ((TYPE_DATASET != $type) && (! ref $data)) {
+               my $pkg = scalar caller;
+
+               my %p : shared;
+
+               if ($data !~ m/^$pkg/) {
+                       $data = $pkg . "::" . $data;
+               }
+
                # TODO: make interval_g available at configuration time
-               $plugins[$type]->{$name} = {
-                               wait_time => 10,
-                               wait_left => 0,
-                               code      => $data,
-               };
+               %p = (
+                       wait_time => 10,
+                       wait_left => 0,
+                       cb_name   => $data,
+               );
+
+               lock @plugins;
+               $plugins[$type]->{$name} = \%p;
        }
        else {
                ERROR ("Collectd::plugin_register: Invalid data.");
@@ -224,6 +254,7 @@ sub plugin_unregister {
                return plugin_unregister_data_set ($name);
        }
        elsif (defined $plugins[$type]) {
+               lock @plugins;
                delete $plugins[$type]->{$name};
        }
        else {