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; # also --$BUSY

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) {
      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 {
         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;

                  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);

               kick_job;
            }
         })
      ); 
   } else {
     kick_job;
   }
}

sub kick_job {
   --$BUSY;

   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;
         kick_job;
      };
      return 1;
   }

   push @QUEUE, $pdu;
   kick_job;

   1
}

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

sub one_event($) {
   AnyEvent->one_event; #d# todo
}

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

   ++$BUSY; # kick_job decrements $BUSY
   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.8
Committed:	Wed Jan 6 10:25:54 2010 UTC (14 years, 5 months ago) by root
Branch:	MAIN
Changes since 1.7:	+25 -24 lines
Log Message:	* empty log message *
#	Content
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	=head1 GLOBAL VARIABLES
49
50	=over 4
51
52	=item $AnyEvent::SNMP::MAX_OUTSTANDING (default: C<50>, dynamic)
53
54	=item AnyEvent::SNMP::set_max_outstanding $new_value
55
56	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	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
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	Increasing C<$MAX_OUTSTANDING> will not automatically use the
80	extra request slots. To increase C<$MAX_OUTSTANDING> and make
81	C<AnyEvent::SNMP> make use of the extra paralellity, call
82	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	Note that you can use L<Net::SNMP::XS> to speed up parsing of responses
90	considerably.
91
92	=item $AnyEvent::SNMP::MIN_RECVQUEUE (default: C<8>)
93
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	=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	our $VERSION = '1.0';
151
152	$Net::SNMP::DISPATCHER = instance Net::SNMP::Dispatcher;
153
154	our $MESSAGE_PROCESSING = $Net::SNMP::Dispatcher::MESSAGE_PROCESSING;
155
156	our $BUSY;
157	our $DONE; # finished all jobs
158	our @TRANSPORT; # fileno => [count, watcher]
159	our @QUEUE;
160	our $MAX_OUTSTANDING = 50;
161	our $MIN_RECVQUEUE = 8;
162	our $MAX_RECVQUEUE = 64;
163
164	sub kick_job; # also --$BUSY
165
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	kick_job;
177	# Inform the command generator about the Message Processing error.
178	$pdu->status_information ($MESSAGE_PROCESSING->error);
179	return;
180	}
181
182	# Actually send the message.
183	if (!defined $msg->send) {
184	$MESSAGE_PROCESSING->msg_handle_delete ($pdu->msg_id)
185	if $pdu->expect_response;
186
187	# A crude attempt to recover from temporary failures.
188	if ($retries-- > 0 && ($!{EAGAIN} \|\| $!{EWOULDBLOCK} \|\| $!{ENOSPC})) {
189	my $retry_w; $retry_w = AE::timer $pdu->timeout, 0, sub {
190	undef $retry_w;
191	_send_pdu ($pdu, $retries);
192	};
193	} else {
194	kick_job;
195	}
196
197	# Inform the command generator about the send() error.
198	$pdu->status_information ($msg->error);
199	return;
200	}
201
202	# Schedule the timeout handler if the message expects a response.
203	if ($pdu->expect_response) {
204	my $transport = $msg->transport;
205	my $fileno = $transport->fileno;
206
207	# register the transport
208	unless ($TRANSPORT[$fileno][0]++) {
209	$TRANSPORT[$fileno][1] = AE::io $transport->socket, 0, sub {
210	for my $count (1..$MAX_RECVQUEUE) { # handle up to this many requests in one go
211	# Create a new Message object to receive the response
212	my ($msg, $error) = Net::SNMP::Message->new (-transport => $transport);
213
214	if (!defined $msg) {
215	die sprintf 'Failed to create Message object [%s]', $error;
216	}
217
218	# Read the message from the Transport Layer
219	if (!defined $msg->recv) {
220	if ($transport->connectionless) {
221	# if we handled very few replies and we have queued work, try
222	# to increase the parallelity as we probably can handle more.
223	if ($count < $MIN_RECVQUEUE && @QUEUE) {
224	++$MAX_OUTSTANDING;
225	kick_job;
226	}
227	} else {
228	# for some reason, connected-oriented transports seem to need this
229	delete $TRANSPORT[$fileno]
230	unless --$TRANSPORT[$fileno][0];
231	}
232
233	$msg->error;
234	return;
235	}
236
237	# For connection-oriented Transport Domains, it is possible to
238	# "recv" an empty buffer if reassembly is required.
239	if (!$msg->length) {
240	return;
241	}
242
243	# Hand the message over to Message Processing.
244	if (!defined $MESSAGE_PROCESSING->prepare_data_elements ($msg)) {
245	$MESSAGE_PROCESSING->error;
246	return;
247	}
248
249	# Set the error if applicable.
250	$msg->error ($MESSAGE_PROCESSING->error) if $MESSAGE_PROCESSING->error;
251
252	# Notify the command generator to process the response.
253	$msg->process_response_pdu;
254
255	# Cancel the timeout.
256	my $rtimeout_w = $msg->timeout_id;
257	if ($$rtimeout_w) {
258	undef $$rtimeout_w;
259
260	kick_job;
261
262	unless (--$TRANSPORT[$fileno][0]) {
263	delete $TRANSPORT[$fileno];
264	return;
265	}
266	}
267	}
268
269	# when we end up here, we successfully handled $MAX_RECVQUEUE
270	# replies in one iteration, so assume we are overloaded
271	# and reduce the amount of parallelity.
272	$MAX_OUTSTANDING = (int $MAX_OUTSTANDING * 0.95) \|\| 1;
273	};
274	}
275
276	$msg->timeout_id (\(my $rtimeout_w =
277	AE::timer $pdu->timeout, 0, sub {
278	my $rtimeout_w = $msg->timeout_id;
279	if ($$rtimeout_w) {
280	undef $$rtimeout_w;
281	delete $TRANSPORT[$fileno]
282	unless --$TRANSPORT[$fileno][0];
283	}
284
285	if ($retries--) {
286	_send_pdu ($pdu, $retries);
287	} else {
288	$MESSAGE_PROCESSING->msg_handle_delete ($pdu->msg_id);
289	$pdu->status_information ("No response from remote host '%s'", $pdu->hostname);
290
291	kick_job;
292	}
293	})
294	);
295	} else {
296	kick_job;
297	}
298	}
299
300	sub kick_job {
301	--$BUSY;
302
303	while ($BUSY < $MAX_OUTSTANDING) {
304	my $pdu = shift @QUEUE
305	or last;
306
307	++$BUSY;
308
309	_send_pdu $pdu, $pdu->retries;
310	}
311
312	$DONE and $DONE->() unless $BUSY;
313	}
314
315	sub send_pdu($$$) {
316	my (undef, $pdu, $delay) = @_;
317
318	# $delay is not very sensibly implemented by AnyEvent::SNMP,
319	# but apparently it is not a very sensible feature.
320	if ($delay > 0) {
321	++$BUSY;
322	my $delay_w; $delay_w = AE::timer $delay, 0, sub {
323	undef $delay_w;
324	push @QUEUE, $pdu;
325	kick_job;
326	};
327	return 1;
328	}
329
330	push @QUEUE, $pdu;
331	kick_job;
332
333	1
334	}
335
336	sub activate($) {
337	while ($BUSY) {
338	$DONE = AE::cv;
339	$DONE->recv;
340	undef $DONE;
341	}
342	}
343
344	sub one_event($) {
345	AnyEvent->one_event; #d# todo
346	}
347
348	sub set_max_outstanding($) {
349	$MAX_OUTSTANDING = $_[0];
350
351	++$BUSY; # kick_job decrements $BUSY
352	kick_job;
353	}
354
355	=head1 SEE ALSO
356
357	L<AnyEvent>, L<Net::SNMP>, L<Net::SNMP::XS>, L<Net::SNMP::EV>.
358
359	=head1 AUTHOR
360
361	Marc Lehmann <schmorp@schmorp.de>
362	http://home.schmorp.de/
363
364	=cut
365
366	1
367