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)

=item AnyEvent::SNMP::set_max_outstanding $new_value

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.

Increasing C<$MAX_OUTSTANDING> will not automatically use the
extra request slots. To increase C<$MAX_OUTSTANDING> and make
C<AnyEvent::SNMP> make use of the extra paralellity, call
C<AnyEvent::SNMP::set_max_outstanding> with the new value, e.g.:

   AnyEvent::SNMP::set_max_outstanding 500;

Although due to the dynamic adjustment, this might have little lasting
effect.

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 = '1.0';

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

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

our $BUSY;
our $DONE; # finished all jobs
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 = AE::timer $pdu->timeout, 0, 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] = AE::io $transport->socket, 0, 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 =
         AE::timer $pdu->timeout, 0, 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;
   }

   $DONE and $DONE->() unless $BUSY;
}

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 = AE::timer $delay, 0, sub {
         undef $delay_w;
         push @QUEUE, $pdu;
         --$BUSY;
         kick_job;
      };
      return 1;
   }

   push @QUEUE, $pdu;
   kick_job;

   1
}

sub activate($) {
   while ($BUSY) {
      $DONE = AE::cv;
      $DONE->recv;
      undef $DONE;
   }
}

sub one_event($) {
   # should not ever be used
   AnyEvent->one_event; #d# todo
}

sub set_max_outstanding($) {
   $MAX_OUTSTANDING = $_[0];
   kick_job;
}

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