File Coverage

blib/lib/SNMP/Query/Asynch.pm

Criterion	Covered	Total	%
statement	10	12	83.3
branch			n/a
condition			n/a
subroutine	4	4	100.0
pod			n/a
total	14	16	87.5

# -*- mode: perl -*-
#
# Copyright (c) 2008 Stephen R. Scaffidi
# All rights reserved.
#
# This program is free software; you may redistribute it and/or modify it
# under the same terms as Perl itself.
#
# Current RCS Info:
#
#       $Id: Asynch.pm 38 2008-03-11 16:48:32Z hercynium $
#  $HeadURL: https://rtg-utilities.svn.sourceforge.net/svnroot/rtg-utilities/SNMP-Asynch/lib/SNMP/Query/Asynch.pm $
#     $Date: 2008-03-11 12:48:32 -0400 (Tue, 11 Mar 2008) $
#   $Author: hercynium $
# $Revision: 38 $

package SNMP::Query::Asynch;

# Pragmas
use strict;
use warnings;

# Standard
use Carp;

# Cpan
use SNMP;

# See chap 17, pg. 404, PBP (Conway 2005)
# use version; our $VERSION = qv('0.1_32');
use version; our $VERSION = qv(sprintf "0.1_%d", q$Revision: 38 $ =~ /: (\d+)/);


# This comes in handy so we don't pass bogus
# parameters to SNMP::Session->new()
use vars qw(@valid_sess_params);
@valid_sess_params = qw(
      DestHost
      Community
      Version
      RemotePort
      Timeout
      Retries
      RetryNoSuch
      SecName
      SecLevel
      SecEngineId
      ContextEngineId
      Context
      AuthProto
      AuthPass
      PrivProto
      PrivPass
      AuthMasterKey
      PrivMasterKey
      AuthLocalizedKey
      PrivLocalizedKey
      VarFormats
      TypeFormats
      UseLongNames
      UseSprintValue
      UseEnums
      UseNumeric
      BestGuess
      NonIncreasing
    );

#----------------------------------------------------------
# Constructor
sub new {
    my $class = shift;
    my $self = bless {}, $class;

    $self->{query_stack}          = [];
    $self->{results}              = [];
    $self->{max_in_flight}        = 10;
    $self->{current_in_flight}    = 0;

    $self->{this_run_issued}      = 0;
    $self->{this_run_finished}    = 0;

    $self->{grand_total_issued}   = 0;
    $self->{grand_total_finished} = 0;

    return $self;
}

#---------------------------------------------------------

# Verifies that the named parameter is a subref
sub _check_param_callback {
    my $self       = shift;
    my $param_name = shift;    # string, hash key
    my $params     = shift;    # hashref
    return 1 unless exists $params->{$param_name};
    croak "Bad parameter for [$param_name] - not a CODE ref"
      if ref $params->{$param_name} ne 'CODE';
    return 1;
}

#---------------------------------------------------------

# TODO Fill in the code and use in the add_XXX() or _make_XXX_query() methods
# Verifies that the named parameter is something the SNMP
# module can use as a VarBind or VarBindList.
sub _check_param_varbinds {
    my $self       = shift;
    my $param_name = shift;    # string, hash key
    my $params     = shift;    # hashref
    return;
}

#---------------------------------------------------------

sub add_getbulk {
    my $self   = shift;
    my $params = shift;

    my $query_stack = $self->{query_stack};

    # These are required for all query ops so make sure they're present.
    my $varbinds = $params->{VarBinds}
      || croak "Bad or missing parameter [VarBinds]";
    my $query_type = $params->{QueryType}
      || croak "Bad or missing parameter [QueryType]";

    # Make sure our callback params are valid.
    $self->_check_param_callback( 'PreCallback',   $params );
    $self->_check_param_callback( 'PostCallback',  $params );

    if ( $query_type eq 'getbulk' ) {
        my $query       = $self->_make_getbulk_query($params);
        my $query_stack = $self->{query_stack};
        push @$query_stack, $query;
    }
    else {
        croak "Attempt to add using unknown query type: $query_type\n";
    }

    return;
}


#---------------------------------------------------------

# NOTE This method has gotten long and complex.
# TODO Refactor, but think carefully.
#
# I see a couple of places where I can create closure-based memory leaks, and
# knowing myself, I could end up doing it inadvertenly at any time. It's code
# like this where some more experienced hackers could help me *big time*
sub _make_getbulk_query {
    my $self = shift;
    my $query_info = shift;

    # These params are required for a getbulk query op.
    my $non_repeaters =
      exists $query_info->{NonRepeaters}
      ? $query_info->{NonRepeaters}
      : croak "Bad or missing parameter [NonRepeaters]";
    my $max_repeaters =
      exists $query_info->{MaxRepeaters}
      ? $query_info->{MaxRepeaters}
      : croak "Bad or missing parameter [MaxRepeaters]";

    # Currently, these are validated in the add() method, so no need here.
    my $preop_callback  = $query_info->{PreCallback};
    my $postop_callback = $query_info->{PostCallback};
    my $batch_callback  = $query_info->{BatchCallback};

    # TODO I may want to add a method to validate
    # and/or transform the VarBinds parameter
    my $varbinds = $query_info->{VarBinds};

    # Parse out the parameters for creating the session.
    # I really think I should be validating them better here...
    # Maybe I need a separate subroutine...
    # TODO write the routine described above.
    my %sess_params;
    $sess_params{$_} = $query_info->{$_}
      for grep { exists $query_info->{$_} } @valid_sess_params;

    my $batch_size = $query_info->{BatchSize};

    return sub {

        # I wonder if this should be before or after the counter increments?
        $preop_callback->() if $preop_callback;

        $self->{current_in_flight}++;
        $self->{this_run_issued}++;
        $self->{grand_total_issued}++;

        my $callback = sub {

            my $bulkwalk_results = shift;

            # Store the results and info about the query for later...
            push @{ $self->{results} }, $query_info, $bulkwalk_results;

            # NOTE should I do this callback here, or somewhere else? hmmmmm....
            $postop_callback->() if $postop_callback;

            # Am I introducing a bug with the increments
            # here and the SNMP::finish() down below?
            $self->{current_in_flight}--;
            $self->{this_run_finished}++;
            $self->{grand_total_finished}++;

            $self->{BatchCallback}()
              if $self->{BatchSize}
                  && ( $self->{this_run_finished} % $self->{BatchSize} == 0 );

            if ( scalar @{ $self->{query_stack} } ) {
                my $next_query = pop @{ $self->{query_stack} };
                return $next_query->();
            }
            $self->{current_in_flight} <= 0 ? return SNMP::finish() : return 1;
        };

        my $sess = SNMP::Session->new(%sess_params);

        my $operation_id = $sess->bulkwalk( $non_repeaters, $max_repeaters,
                                            $varbinds, [$callback] );

        # Why, yes. I am being sneaky here. Since $query_info is referenced
        # within the $callback closure, any data stored in the hash it points
        # to should be available, regardless of whether is was stored before
        # or after the closure definition.
        $query_info->{OperationId} = $operation_id;

        return; # No need to return anything, AFAICT.
    };
}

#---------------------------------------------------------



sub current_in_flight    { return shift->{current_in_flight} }
sub this_run_issued      { return shift->{this_run_issued} }
sub this_run_finished    { return shift->{this_run_finished} }
sub grand_total_issued   { return shift->{grand_total_issued} }
sub grand_total_finished { return shift->{grand_total_finished} }

sub shuffle { } # TODO Implement with List::Util::shuffle

#---------------------------------------------------------

sub execute {
    my $self   = shift;
    my $params = shift;

    # The KeepLast option can come in handy if, for example, another
    # thread or process is working on the contents of the results
    # array from a previous execution and may not finish before the
    # next execution.
    @{ $self->{results} } = ()
      unless (

        # I'll make my OWN idioms from now on, HAHA! (You can explicitly
        # set keeplast or it will use the object's default)
        defined $params->{KeepLast} ? $params->{KeepLast} : $self->{KeepLast}
      );

    # Install 'batch' callback if applicable...
    $self->{BatchCallback} = $params->{BatchCallback}
        if $self->_check_param_callback( 'BatchCallback', $params );
    $self->{BatchSize} = $params->{BatchSize};


    # Determine our maximum concurrency level for this run
    my $max_in_flight = $params->{InFlight} || $self->{max_in_flight};

    # Make a copy of the stack in case we want to run the same query
    my $query_stack_ref  = $self->{query_stack};
    my @query_stack_copy = @{ $self->{query_stack} };

    # Set some counters
    $self->{current_in_flight} = 0;
    $self->{this_run_issued}   = 0;
    $self->{this_run_finished} = 0;

    # Begin issuing operations
    while ( scalar @$query_stack_ref ) {
        my $query = pop @$query_stack_ref;
        $query->();
        last if $self->{current_in_flight} >= $max_in_flight;
    }

    # Wait for the ops to complete, or time-out (if specified)
    $params->{MasterTimeout}
      ? SNMP::MainLoop( $params->{MasterTimeout}, &SNMP::finish() )
      : SNMP::MainLoop();

    # Reset the stack for the next run.
    $self->{query_stack} = \@query_stack_copy;

    return $self->get_results_ref();
}

#---------------------------------------------------------

#  Returns a reference to the results array from executing the query.
sub get_results_ref {
    return shift->{results};
}

1;
__END__