cvsroot/AnyEvent-SNMP/SNMP.pm

=head1 NAME

AnyEvent::SNMP - adaptor to integrate Net::SNMP into Anyevent.

=head1 SYNOPSIS

 use AnyEvent::SNMP;
 use Net::SNMP;

 # just use Net::SNMP and AnyEvent as you like:

 # use a condvar to transfer results, this is
 # just an example, you can use a naked callback as well.
 my $cv = AnyEvent->condvar;

 # ... start non-blocking snmp request(s)...
 Net::SNMP->session (-hostname => "127.0.0.1",
                     -community => "public",
                     -nonblocking => 1)
          ->get_request (-callback => sub { $cv->send (@_) });

 # ... do something else until the result is required
 my @result = $cv->wait;

=head1 DESCRIPTION

This module implements an alternative "event dispatcher" for Net::SNMP,
using AnyEvent as a backend.

This integrates Net::SNMP into AnyEvent: You can make non-blocking
Net::SNMP calls and as long as other parts of your program also use
AnyEvent (or some event loop supported by AnyEvent), they will run in
parallel.

Also, the Net::SNMP scheduler is very inefficient with respect to both CPU
and memory usage. Most AnyEvent backends (including the pure-perl backend)
fare much better than the Net::SNMP dispatcher.

A potential disadvantage is that replacing the dispatcher is not at all
a documented thing to do, so future changes in Net::SNP might break this
module (or the many similar ones).

This module does not export anything and does not require you to do
anything special apart from loading it I<before doing any non-blocking
requests with Net::SNMP>. It is recommended but not required to load this
module before C<Net::SNMP>.

=head1 GLOBAL VARIABLES

=over 4

=item $AnyEvent::SNMP::MAX_OUTSTANDING (default: C<50>, dynamic)

Use this package variable to restrict the number of outstanding SNMP
requests at any point in time.

Net::SNMP is very fast at creating and sending SNMP requests, but much
slower at parsing (big, bulk) responses. This makes it easy to request a
lot of data that can take many seconds to parse.

In the best case, this can lead to unnecessary delays (and even time-outs,
as the data has been received but not yet processed) and in the worst
case, this can lead to packet loss, when the receive queue overflows and
the kernel can no longer accept new packets.

To avoid this, you can (and should) limit the number of outstanding requests
to a number low enough so that parsing time doesn't introduce noticable delays.

Unfortunately, this number depends not only on processing speed and load
of the machine running Net::SNMP, but also on the network latency and the
speed of your SNMP agents.

AnyEvent::SNMP tries to dynamically adjust this number dynamically upwards
and downwards.

Note that you can use L<Net::SNMP::XS> to speed up parsing of responses
considerably.

=item $AnyEvent::SNMP::MIN_RECVQUEUE (default: C<8>)

=item $AnyEvent::SNMP::MAX_RECVQUEUE (default: C<64>)

These values specify the minimum and maximum receive queue length (in
units of one response packet).

When AnyEvent::SNMP handles $MAX_RECVQUEUE or more packets per iteration
it will reduce $MAX_OUTSTANDING. If it handles less than $MIN_RECVQUEUE,
it increases $MAX_OUTSTANDING.

This has the result of adjusting the number of outstanding requests so that
the recv queue is between the minimum and maximu, usually.

This algorithm works reasonably well as long as the responses, response
latencies and processing times are the same size per packet on average.

=back

=head1 COMPATIBILITY

This module may be used as a drop in replacement for the
Net::SNMP::Dispatcher in existing programs. You can still call
C<snmp_dispatcher> to start the event-loop, but then you loose the benefit
of mixing Net::SNMP events with other events.

   use AnyEvent::SNMP;
   use Net::SNMP;

   # just use Net::SNMP as before

   # ... start non-blocking snmp request(s)...
   Net::SNMP->session (
         -hostname    => "127.0.0.1",
         -community   => "public",
         -nonblocking => 1,
      )->get_request (-callback => sub { ... });

   snmp_dispatcher;

=cut

package AnyEvent::SNMP;

no warnings;
use strict qw(subs vars);

# it is possible to do this without loading
# Net::SNMP::Dispatcher, but much more awkward.
use Net::SNMP::Dispatcher;

sub Net::SNMP::Dispatcher::instance {
   AnyEvent::SNMP::
}

use Net::SNMP ();
use AnyEvent ();

our $VERSION = '0.2';

$Net::SNMP::DISPATCHER = instance Net::SNMP::Dispatcher;

our $MESSAGE_PROCESSING = $Net::SNMP::Dispatcher::MESSAGE_PROCESSING;

# avoid the method call
my $timer = sub { shift->timer (@_) };
AnyEvent::post_detect { $timer = AnyEvent->can ("timer") };

our $BUSY;
our @TRANSPORT; # fileno => [count, watcher]
our @QUEUE;
our $MAX_OUTSTANDING = 50;
our $MIN_RECVQUEUE   =  8;
our $MAX_RECVQUEUE   = 64;

sub kick_job;

sub _send_pdu {
   my ($pdu, $retries) = @_;

   # mostly copied from Net::SNMP::Dispatch

   # Pass the PDU to Message Processing so that it can
   # create the new outgoing message.
   my $msg = $MESSAGE_PROCESSING->prepare_outgoing_msg ($pdu);

   if (!defined $msg) {
      --$BUSY;
      kick_job;
      # Inform the command generator about the Message Processing error.
      $pdu->status_information ($MESSAGE_PROCESSING->error);
      return; 
   }

   # Actually send the message.
   if (!defined $msg->send) {
      $MESSAGE_PROCESSING->msg_handle_delete ($pdu->msg_id)
         if $pdu->expect_response;

      # A crude attempt to recover from temporary failures.
      if ($retries-- > 0 && ($!{EAGAIN} || $!{EWOULDBLOCK} || $!{ENOSPC})) {
         my $retry_w; $retry_w = AnyEvent->$timer (after => $pdu->timeout, cb => sub {
            undef $retry_w;
            _send_pdu ($pdu, $retries);
         });
      } else {
         --$BUSY;
         kick_job;
      }

      # Inform the command generator about the send() error.
      $pdu->status_information ($msg->error);
      return;
   }

   # Schedule the timeout handler if the message expects a response.
   if ($pdu->expect_response) {
      my $transport = $msg->transport;
      my $fileno    = $transport->fileno;

      # register the transport
      unless ($TRANSPORT[$fileno][0]++) {
         $TRANSPORT[$fileno][1] = AnyEvent->io (fh => $transport->socket, poll => 'r', cb => sub {
            for my $count (1..$MAX_RECVQUEUE) { # handle up to this many requests in one go
               # Create a new Message object to receive the response
               my ($msg, $error) = Net::SNMP::Message->new (-transport => $transport);

               if (!defined $msg) {
                  die sprintf 'Failed to create Message object [%s]', $error;
               }

               # Read the message from the Transport Layer
               if (!defined $msg->recv) {
                  if ($transport->connectionless) {
                     # if we handled very few replies and we have queued work, try
                     # to increase the parallelity as we probably can handle more.
                     if ($count < $MIN_RECVQUEUE && @QUEUE) {
                        ++$MAX_OUTSTANDING;
                        kick_job;
                     }
                  } else {
                     # for some reason, connected-oriented transports seem to need this
                     delete $TRANSPORT[$fileno]
                        unless --$TRANSPORT[$fileno][0];
                  }

                  $msg->error;
                  return;
               }

               # For connection-oriented Transport Domains, it is possible to
               # "recv" an empty buffer if reassembly is required.
               if (!$msg->length) {
                  return;
               }

               # Hand the message over to Message Processing.
               if (!defined $MESSAGE_PROCESSING->prepare_data_elements ($msg)) {
                  $MESSAGE_PROCESSING->error;
                  return;
               }

               # Set the error if applicable.
               $msg->error ($MESSAGE_PROCESSING->error) if $MESSAGE_PROCESSING->error;

               # Notify the command generator to process the response.
               $msg->process_response_pdu; 

               # Cancel the timeout.
               my $rtimeout_w = $msg->timeout_id;
               if ($$rtimeout_w) {
                  undef $$rtimeout_w;

                  --$BUSY;
                  kick_job;

                  unless (--$TRANSPORT[$fileno][0]) {
                     delete $TRANSPORT[$fileno];
                     return;
                  }
               }
            }

            # when we end up here, we successfully handled $MAX_RECVQUEUE
            # replies in one iteration, so assume we are overloaded
            # and reduce the amount of parallelity.
            $MAX_OUTSTANDING = (int $MAX_OUTSTANDING * 0.95) || 1;
         });
      }

      $msg->timeout_id (\(my $rtimeout_w =
         AnyEvent->$timer (after => $pdu->timeout, cb => sub {
            my $rtimeout_w = $msg->timeout_id;
            if ($$rtimeout_w) {
               undef $$rtimeout_w;
               delete $TRANSPORT[$fileno]
                  unless --$TRANSPORT[$fileno][0];
            }

            if ($retries--) {
               _send_pdu ($pdu, $retries);
            } else {
               $MESSAGE_PROCESSING->msg_handle_delete ($pdu->msg_id);
               $pdu->status_information ("No response from remote host '%s'", $pdu->hostname);

               --$BUSY;
               kick_job;
            }
         })
      )); 
   } else {
     --$BUSY;
     kick_job;
   }
}

sub kick_job {
   while ($BUSY < $MAX_OUTSTANDING) {
      my $pdu = shift @QUEUE
         or last;

      ++$BUSY;

      _send_pdu $pdu, $pdu->retries;
   }
}
sub send_pdu($$$) {
   my (undef, $pdu, $delay) = @_;

   # $delay is not very sensibly implemented by AnyEvent::SNMP,
   # but apparently it is not a very sensible feature.
   if ($delay > 0) {
      ++$BUSY;
      my $delay_w; $delay_w = AnyEvent->$timer (after => $delay, cb => sub {
         undef $delay_w;
         --$BUSY;
         push @QUEUE, $pdu;
         kick_job;
      });
      return 1;
   }

   push @QUEUE, $pdu;
   kick_job;

   1
}

sub activate($) {
   AnyEvent->one_event while $BUSY;
}

sub one_event($) {
   AnyEvent->one_event;
}

=head1 SEE ALSO

L<AnyEvent>, L<Net::SNMP>, L<Net::SNMP::XS>, L<Net::SNMP::EV>.

=head1 AUTHOR

 Marc Lehmann <schmorp@schmorp.de>
 http://home.schmorp.de/

=cut

1

Revision:	1.5
Committed:	Wed Apr 22 12:31:20 2009 UTC (17 years, 5 months ago) by root
Branch:	MAIN
Changes since 1.4:	+13 -12 lines
Log Message:	* empty log message *
#	User	Rev	Content
1	root	1.1	=head1 NAME
2
3			AnyEvent::SNMP - adaptor to integrate Net::SNMP into Anyevent.
4
5			=head1 SYNOPSIS
6
7			use AnyEvent::SNMP;
8			use Net::SNMP;
9
10			# just use Net::SNMP and AnyEvent as you like:
11
12			# use a condvar to transfer results, this is
13			# just an example, you can use a naked callback as well.
14			my $cv = AnyEvent->condvar;
15
16			# ... start non-blocking snmp request(s)...
17			Net::SNMP->session (-hostname => "127.0.0.1",
18			-community => "public",
19			-nonblocking => 1)
20			->get_request (-callback => sub { $cv->send (@_) });
21
22			# ... do something else until the result is required
23			my @result = $cv->wait;
24
25			=head1 DESCRIPTION
26
27			This module implements an alternative "event dispatcher" for Net::SNMP,
28			using AnyEvent as a backend.
29
30			This integrates Net::SNMP into AnyEvent: You can make non-blocking
31			Net::SNMP calls and as long as other parts of your program also use
32			AnyEvent (or some event loop supported by AnyEvent), they will run in
33			parallel.
34
35			Also, the Net::SNMP scheduler is very inefficient with respect to both CPU
36			and memory usage. Most AnyEvent backends (including the pure-perl backend)
37			fare much better than the Net::SNMP dispatcher.
38
39			A potential disadvantage is that replacing the dispatcher is not at all
40			a documented thing to do, so future changes in Net::SNP might break this
41			module (or the many similar ones).
42
43			This module does not export anything and does not require you to do
44			anything special apart from loading it I<before doing any non-blocking
45			requests with Net::SNMP>. It is recommended but not required to load this
46			module before C<Net::SNMP>.
47
48	root	1.3	=head1 GLOBAL VARIABLES
49
50			=over 4
51
52			=item $AnyEvent::SNMP::MAX_OUTSTANDING (default: C<50>, dynamic)
53
54			Use this package variable to restrict the number of outstanding SNMP
55			requests at any point in time.
56
57			Net::SNMP is very fast at creating and sending SNMP requests, but much
58			slower at parsing (big, bulk) responses. This makes it easy to request a
59			lot of data that can take many seconds to parse.
60
61			In the best case, this can lead to unnecessary delays (and even time-outs,
62			as the data has been received but not yet processed) and in the worst
63			case, this can lead to packet loss, when the receive queue overflows and
64			the kernel can no longer accept new packets.
65
66			To avoid this, you can (and should) limit the number of outstanding requests
67			to a number low enough so that parsing time doesn't introduce noticable delays.
68
69			Unfortunately, this number depends not only on processing speed and load
70			of the machine running Net::SNMP, but also on the network latency and the
71			speed of your SNMP agents.
72
73			AnyEvent::SNMP tries to dynamically adjust this number dynamically upwards
74			and downwards.
75
76			Note that you can use L<Net::SNMP::XS> to speed up parsing of responses
77			considerably.
78
79	root	1.5	=item $AnyEvent::SNMP::MIN_RECVQUEUE (default: C<8>)
80	root	1.3
81			=item $AnyEvent::SNMP::MAX_RECVQUEUE (default: C<64>)
82
83			These values specify the minimum and maximum receive queue length (in
84			units of one response packet).
85
86			When AnyEvent::SNMP handles $MAX_RECVQUEUE or more packets per iteration
87			it will reduce $MAX_OUTSTANDING. If it handles less than $MIN_RECVQUEUE,
88			it increases $MAX_OUTSTANDING.
89
90			This has the result of adjusting the number of outstanding requests so that
91			the recv queue is between the minimum and maximu, usually.
92
93			This algorithm works reasonably well as long as the responses, response
94			latencies and processing times are the same size per packet on average.
95
96			=back
97
98			=head1 COMPATIBILITY
99
100			This module may be used as a drop in replacement for the
101			Net::SNMP::Dispatcher in existing programs. You can still call
102			C<snmp_dispatcher> to start the event-loop, but then you loose the benefit
103			of mixing Net::SNMP events with other events.
104
105			use AnyEvent::SNMP;
106			use Net::SNMP;
107
108			# just use Net::SNMP as before
109
110			# ... start non-blocking snmp request(s)...
111			Net::SNMP->session (
112			-hostname => "127.0.0.1",
113			-community => "public",
114			-nonblocking => 1,
115			)->get_request (-callback => sub { ... });
116
117			snmp_dispatcher;
118
119	root	1.1	=cut
120
121			package AnyEvent::SNMP;
122
123			no warnings;
124			use strict qw(subs vars);
125
126			# it is possible to do this without loading
127			# Net::SNMP::Dispatcher, but much more awkward.
128			use Net::SNMP::Dispatcher;
129
130			sub Net::SNMP::Dispatcher::instance {
131			AnyEvent::SNMP::
132			}
133
134			use Net::SNMP ();
135			use AnyEvent ();
136
137	root	1.3	our $VERSION = '0.2';
138	root	1.1
139			$Net::SNMP::DISPATCHER = instance Net::SNMP::Dispatcher;
140
141			our $MESSAGE_PROCESSING = $Net::SNMP::Dispatcher::MESSAGE_PROCESSING;
142
143			# avoid the method call
144			my $timer = sub { shift->timer (@_) };
145			AnyEvent::post_detect { $timer = AnyEvent->can ("timer") };
146
147			our $BUSY;
148	root	1.5	our @TRANSPORT; # fileno => [count, watcher]
149	root	1.3	our @QUEUE;
150			our $MAX_OUTSTANDING = 50;
151	root	1.5	our $MIN_RECVQUEUE = 8;
152	root	1.3	our $MAX_RECVQUEUE = 64;
153
154			sub kick_job;
155	root	1.1
156			sub _send_pdu {
157			my ($pdu, $retries) = @_;
158
159			# mostly copied from Net::SNMP::Dispatch
160
161			# Pass the PDU to Message Processing so that it can
162			# create the new outgoing message.
163			my $msg = $MESSAGE_PROCESSING->prepare_outgoing_msg ($pdu);
164
165			if (!defined $msg) {
166			--$BUSY;
167	root	1.3	kick_job;
168	root	1.1	# Inform the command generator about the Message Processing error.
169			$pdu->status_information ($MESSAGE_PROCESSING->error);
170			return;
171			}
172
173			# Actually send the message.
174			if (!defined $msg->send) {
175			$MESSAGE_PROCESSING->msg_handle_delete ($pdu->msg_id)
176			if $pdu->expect_response;
177
178			# A crude attempt to recover from temporary failures.
179			if ($retries-- > 0 && ($!{EAGAIN} \|\| $!{EWOULDBLOCK} \|\| $!{ENOSPC})) {
180			my $retry_w; $retry_w = AnyEvent->$timer (after => $pdu->timeout, cb => sub {
181			undef $retry_w;
182			_send_pdu ($pdu, $retries);
183			});
184			} else {
185			--$BUSY;
186	root	1.3	kick_job;
187	root	1.1	}
188
189			# Inform the command generator about the send() error.
190			$pdu->status_information ($msg->error);
191			return;
192			}
193
194			# Schedule the timeout handler if the message expects a response.
195			if ($pdu->expect_response) {
196			my $transport = $msg->transport;
197	root	1.5	my $fileno = $transport->fileno;
198	root	1.1
199			# register the transport
200	root	1.5	unless ($TRANSPORT[$fileno][0]++) {
201			$TRANSPORT[$fileno][1] = AnyEvent->io (fh => $transport->socket, poll => 'r', cb => sub {
202	root	1.3	for my $count (1..$MAX_RECVQUEUE) { # handle up to this many requests in one go
203			# Create a new Message object to receive the response
204			my ($msg, $error) = Net::SNMP::Message->new (-transport => $transport);
205
206			if (!defined $msg) {
207			die sprintf 'Failed to create Message object [%s]', $error;
208			}
209	root	1.1
210	root	1.3	# Read the message from the Transport Layer
211			if (!defined $msg->recv) {
212			if ($transport->connectionless) {
213	root	1.4	# if we handled very few replies and we have queued work, try
214			# to increase the parallelity as we probably can handle more.
215	root	1.3	if ($count < $MIN_RECVQUEUE && @QUEUE) {
216			++$MAX_OUTSTANDING;
217			kick_job;
218			}
219			} else {
220			# for some reason, connected-oriented transports seem to need this
221	root	1.5	delete $TRANSPORT[$fileno]
222			unless --$TRANSPORT[$fileno][0];
223	root	1.3	}
224	root	1.1
225	root	1.3	$msg->error;
226			return;
227	root	1.1	}
228
229	root	1.3	# For connection-oriented Transport Domains, it is possible to
230			# "recv" an empty buffer if reassembly is required.
231			if (!$msg->length) {
232			return;
233			}
234	root	1.1
235	root	1.3	# Hand the message over to Message Processing.
236			if (!defined $MESSAGE_PROCESSING->prepare_data_elements ($msg)) {
237			$MESSAGE_PROCESSING->error;
238			return;
239			}
240	root	1.1
241	root	1.3	# Set the error if applicable.
242			$msg->error ($MESSAGE_PROCESSING->error) if $MESSAGE_PROCESSING->error;
243	root	1.1
244	root	1.3	# Notify the command generator to process the response.
245			$msg->process_response_pdu;
246	root	1.1
247	root	1.3	# Cancel the timeout.
248			my $rtimeout_w = $msg->timeout_id;
249			if ($$rtimeout_w) {
250			undef $$rtimeout_w;
251
252			--$BUSY;
253			kick_job;
254
255	root	1.5	unless (--$TRANSPORT[$fileno][0]) {
256			delete $TRANSPORT[$fileno];
257	root	1.3	return;
258			}
259			}
260	root	1.1	}
261
262	root	1.4	# when we end up here, we successfully handled $MAX_RECVQUEUE
263			# replies in one iteration, so assume we are overloaded
264			# and reduce the amount of parallelity.
265	root	1.5	$MAX_OUTSTANDING = (int $MAX_OUTSTANDING * 0.95) \|\| 1;
266	root	1.1	});
267			}
268
269			$msg->timeout_id (\(my $rtimeout_w =
270			AnyEvent->$timer (after => $pdu->timeout, cb => sub {
271			my $rtimeout_w = $msg->timeout_id;
272			if ($$rtimeout_w) {
273			undef $$rtimeout_w;
274	root	1.5	delete $TRANSPORT[$fileno]
275			unless --$TRANSPORT[$fileno][0];
276	root	1.1	}
277
278			if ($retries--) {
279			_send_pdu ($pdu, $retries);
280			} else {
281			$MESSAGE_PROCESSING->msg_handle_delete ($pdu->msg_id);
282			$pdu->status_information ("No response from remote host '%s'", $pdu->hostname);
283	root	1.3
284			--$BUSY;
285			kick_job;
286	root	1.1	}
287			})
288			));
289			} else {
290			--$BUSY;
291	root	1.3	kick_job;
292	root	1.1	}
293			}
294
295	root	1.3	sub kick_job {
296			while ($BUSY < $MAX_OUTSTANDING) {
297			my $pdu = shift @QUEUE
298			or last;
299
300			++$BUSY;
301
302			_send_pdu $pdu, $pdu->retries;
303			}
304			}
305	root	1.1	sub send_pdu($$$) {
306			my (undef, $pdu, $delay) = @_;
307
308	root	1.3	# $delay is not very sensibly implemented by AnyEvent::SNMP,
309			# but apparently it is not a very sensible feature.
310	root	1.1	if ($delay > 0) {
311	root	1.3	++$BUSY;
312	root	1.1	my $delay_w; $delay_w = AnyEvent->$timer (after => $delay, cb => sub {
313			undef $delay_w;
314	root	1.3	--$BUSY;
315			push @QUEUE, $pdu;
316			kick_job;
317	root	1.1	});
318			return 1;
319			}
320
321	root	1.3	push @QUEUE, $pdu;
322			kick_job;
323
324	root	1.1	1
325			}
326
327			sub activate($) {
328			AnyEvent->one_event while $BUSY;
329			}
330
331			sub one_event($) {
332	root	1.3	AnyEvent->one_event;
333	root	1.1	}
334
335			=head1 SEE ALSO
336
337	root	1.3	L<AnyEvent>, L<Net::SNMP>, L<Net::SNMP::XS>, L<Net::SNMP::EV>.
338	root	1.1
339			=head1 AUTHOR
340
341			Marc Lehmann <schmorp@schmorp.de>
342			http://home.schmorp.de/
343
344			=cut
345
346			1
347