ViewVC Help
View File | Revision Log | Show Annotations | Download File
/cvs/AnyEvent-FastPing/FastPing.xs
(Generate patch)

Comparing AnyEvent-FastPing/FastPing.xs (file contents):
Revision 1.1 by root, Sun Apr 27 15:43:51 2008 UTC vs.
Revision 1.11 by root, Tue Feb 1 04:06:24 2011 UTC

1#define _POSIX_C_SOURCE 199309 1#if defined(__linux) || defined(__FreeBSD__) || defined(__OpenBSD__) || defined(__NetBSD__) || defined(__CYGWIN__)
2#define _GNU_SOURCE 1
3
4#define IPV6 1 // if you get compilation problems try to disable IPv6 2# define ENABLE_IPV6 1 // if you get compilation problems try to disable IPv6
3#else
4# define ENABLE_IPV6 0
5#endif
5 6
6#include "EXTERN.h" 7#include "EXTERN.h"
7#include "perl.h" 8#include "perl.h"
8#include "XSUB.h" 9#include "XSUB.h"
9 10
10#include <pthread.h> 11#include <pthread.h>
11 12
12#include <math.h>
13#include <stdio.h> 13#include <stdio.h>
14#include <stdlib.h> 14#include <stdlib.h>
15#include <string.h> 15#include <string.h>
16 16
17#include <time.h> 17#include <time.h>
… …
29#include <arpa/inet.h> 29#include <arpa/inet.h>
30 30
31#ifdef __linux 31#ifdef __linux
32# include <linux/icmp.h> 32# include <linux/icmp.h>
33#endif 33#endif
34#if IPV6 34#if ENABLE_IPV6 && !defined (__CYGWIN__)
35# include <netinet/icmp6.h> 35# include <netinet/icmp6.h>
36#endif 36#endif
37 37
38#define ICMP4_ECHO 8 38#define ICMP4_ECHO 8
39#define ICMP4_ECHO_REPLY 0 39#define ICMP4_ECHO_REPLY 0
40#define ICMP6_ECHO 128 40#define ICMP6_ECHO 128
41#define ICMP6_ECHO_REPLY 129 41#define ICMP6_ECHO_REPLY 129
42 42
43#define DRAIN_INTERVAL .000001 // how long to wait when sendto returns ENOBUFS, in seconds 43#define DRAIN_INTERVAL 1e-6 // how long to wait when sendto returns ENOBUFS, in seconds
44#define MIN_INTERVAL .000001 // minimum packet send interval, in seconds 44#define MIN_INTERVAL 1e-6 // minimum packet send interval, in seconds
45 45
46#define HDR_SIZE_IP4 20 46#define HDR_SIZE_IP4 20
47#define HDR_SIZE_IP6 48 47#define HDR_SIZE_IP6 48
48 48
49//TODO: xread/xwrite for atomicity? we currently rely on the fact that the pip biffersize divides exactly by pointer sizes 49static int thr_res[2]; // worker thread finished status
50static int icmp4_fd = -1;
51static int icmp6_fd = -1;
50 52
51typedef uint8_t addr_t[16]; 53/*****************************************************************************/
52 54
53typedef double tstamp; 55typedef double tstamp;
54 56
55tstamp 57static tstamp
56NOW () 58NOW (void)
57{ 59{
58 struct timeval tv; 60 struct timeval tv;
61
59 gettimeofday (&tv, 0); 62 gettimeofday (&tv, 0);
63
60 return tv.tv_sec + tv.tv_usec * 0.000001; 64 return tv.tv_sec + tv.tv_usec * 1e-6;
61} 65}
62 66
67static void
68ssleep (tstamp wait)
69{
70#if defined (__SVR4) && defined (__sun)
71 struct timeval tv;
72
73 tv.tv_sec = wait;
74 tv.tv_usec = (wait - tv.tv_sec) * 1e6;
75
76 select (0, 0, 0, 0, &tv);
77#elif defined(_WIN32)
78 Sleep ((unsigned long)(delay * 1e3));
79#else
80 struct timespec ts;
81
82 ts.tv_sec = wait;
83 ts.tv_nsec = (wait - ts.tv_sec) * 1e9;
84
85 nanosleep (&ts, 0);
86#endif
87}
88
89/*****************************************************************************/
90
63typedef struct { 91typedef struct
64 int family; 92{
65 addr_t lo, hi;
66 double interval;
67 tstamp next;
68} RANGE;
69
70typedef struct {
71 SV *id;
72 double interval;
73 int nranges;
74 RANGE *ranges;
75 uint32_t payload;
76} REQ;
77
78typedef struct {
79 uint8_t version_ihl; 93 uint8_t version_ihl;
80 uint8_t tos; 94 uint8_t tos;
81 uint16_t tot_len; 95 uint16_t tot_len;
82 96
83 uint16_t id; 97 uint16_t id;
… …
89 103
90 uint32_t src; 104 uint32_t src;
91 uint32_t dst; 105 uint32_t dst;
92} IP4HDR; 106} IP4HDR;
93 107
108/*****************************************************************************/
109
110typedef uint8_t addr_tt[16];
111
94typedef struct { 112typedef struct
95 uint8_t version; 113{
96 uint8_t x1, x2, x3; 114 tstamp next;
115 double interval;
116 int addrlen;
97 117
98 uint16_t payload_len; 118 addr_tt lo, hi; /* only if !addrcnt */
99 uint8_t nxt_hdr;
100 uint8_t hop_limit;
101 119
102 uint8_t src[16]; 120 int addrcnt;
103 uint8_t dst[16]; 121 /* addrcnt addresses follow */
104} IP6HDR; 122} RANGE;
105 123
106#define MAGIC 0xca4c
107
108static uint16_t magic;
109
110typedef struct { 124typedef struct
125{
126 RANGE **ranges;
127 int rangecnt, rangemax;
128
129 tstamp interval;
130 tstamp maxrtt;
131 uint16_t magic1;
132 uint16_t magic2;
133 uint16_t magic3;
134
135 int id;
136
137 AV *recvq; /* receive queue */
138 int nextrecv;
139 SV *recvcb;
140
141 pthread_t thrid;
142 int running;
143} PINGER;
144
145static PINGER **pingers;
146static int *pingerfree; /* freelist next */
147static int pingercnt;
148static int pingermax;
149static int firstfree = -1;
150static int firstrecv = -1;
151
152/*****************************************************************************/
153
154typedef struct
155{
111 uint8_t type, code; 156 uint8_t type, code;
112 uint16_t cksum; 157 uint16_t cksum;
158
113 uint16_t id, seq; 159 uint16_t id, seq;
160
161 uint16_t pinger;
162 uint16_t magic;
163
164 uint32_t stamp_hi;
114 uint32_t payload; 165 uint32_t stamp_lo;
115 tstamp stamp; // be careful when accessing this
116} PKT; 166} PKT;
117 167
118static pthread_t pthrid;
119static int thr_send[2]; // send to worker
120static int thr_recv[2]; // receive from worker
121
122static int icmp4_fd, icmp6_fd;
123
124static AV *cbs;
125
126static uint16_t 168static int
127icmp_cksum (void *data, unsigned int len) 169pkt_is_valid_for (PKT *pkt, PINGER *pinger)
128{ 170{
129 register int sum = 0; 171 return pkt->id == pinger->magic1
130 uint16_t *wp; 172 && pkt->seq == pinger->magic2
173 && pkt->magic == pinger->magic3;
174}
131 175
132 assert (~len & 1); 176static void
177ts_to_pkt (PKT *pkt, tstamp ts)
178{
179 /* move 12 bits of seconds into the 32 bit fractional part */
180 /* leaving 20 bits subsecond resolution and 44 bits of integers */
181 /* (of which 32 are typically usable) */
182 ts *= 1. / 4096.;
133 183
134 for (wp = (uint16_t *)data; len; wp++, len -= 2) 184 pkt->stamp_hi = ts;
135 sum += *wp; 185 pkt->stamp_lo = (ts - pkt->stamp_hi) * 4294967296.;
186}
187
188static tstamp
189pkt_to_ts (PKT *pkt)
190{
191 return pkt->stamp_hi * 4096.
192 + pkt->stamp_lo * (4096. / 4294967296.);
193}
194
195static void
196pkt_cksum (PKT *pkt)
197{
198 uint_fast32_t sum = -pkt->cksum;
199 uint32_t *wp = (uint32_t *)pkt;
200 int len = sizeof (*pkt) / 4;
201
202 do
203 {
204 uint_fast32_t w = *(volatile uint32_t *)wp++;
205 sum += (w & 0xffff) + (w >> 16);
206 }
207 while (len--);
136 208
137 sum = (sum >> 16) + (sum & 0xffff); /* add high 16 to low 16 */ 209 sum = (sum >> 16) + (sum & 0xffff); /* add high 16 to low 16 */
138 sum += sum >> 16; /* add carry */ 210 sum += sum >> 16; /* add carry */
139 211
212 pkt->cksum = ~sum;
213}
214
215/*****************************************************************************/
216
217static void
218range_free (RANGE *self)
219{
220 free (self);
221}
222
223/* like sendto, but retries on failure */
224static void
225xsendto (int fd, void *buf, size_t len, int flags, void *sa, int salen)
226{
227 tstamp wait = DRAIN_INTERVAL / 2.;
228
229 while (sendto (fd, buf, len, flags, sa, salen) < 0 && errno == ENOBUFS)
230 ssleep (wait *= 2.);
231}
232
233// ping current address, return true and increment if more to ping
234static int
235range_send_ping (RANGE *self, PKT *pkt)
236{
237 // send ping
238 uint8_t *addr;
239
240 if (self->addrcnt)
241 addr = (self->addrcnt - 1) * self->addrlen + (uint8_t *)(self + 1);
242 else
243 addr = sizeof (addr_tt) - self->addrlen + self->lo;
244
245 pkt->cksum = 0;
246
247 if (self->addrlen == 4)
248 {
249 struct sockaddr_in sa;
250
251 pkt->type = ICMP4_ECHO;
252 pkt_cksum (pkt);
253
254 sa.sin_family = AF_INET;
255 sa.sin_port = 0;
256
257 memcpy (&sa.sin_addr, addr, sizeof (sa.sin_addr));
258
259 xsendto (icmp4_fd, pkt, sizeof (*pkt), 0, &sa, sizeof (sa));
260 }
261 else
262 {
263#if ENABLE_IPV6
264 struct sockaddr_in6 sa;
265
266 pkt->type = ICMP6_ECHO;
267
268 sa.sin6_family = AF_INET6;
269 sa.sin6_port = 0;
270 sa.sin6_flowinfo = 0;
271 sa.sin6_scope_id = 0;
272
273 memcpy (&sa.sin6_addr, addr, sizeof (sa.sin6_addr));
274
275 xsendto (icmp6_fd, pkt, sizeof (*pkt), 0, &sa, sizeof (sa));
276#endif
277 }
278
279 // see if we have any more addresses
280 if (self->addrcnt)
281 {
282 if (!--self->addrcnt)
283 return 0;
284 }
285 else
286 {
287 if (!memcmp (&self->lo, &self->hi, sizeof (addr_tt)))
288 return 0;
289
290 // increment self->lo
291 {
292 int len = sizeof (addr_tt) - 1;
293
294 while (!++self->lo [len])
295 --len;
296 }
297 }
298
140 return ~sum; 299 return 1;
141} 300}
142 301
302/*****************************************************************************/
303
304static void
305downheap (PINGER *self)
306{
307 RANGE *elem = self->ranges [0]; /* always exists */
308 int Nm1 = self->rangecnt - 1;
309 int j;
310 int k;
311
312 for (k = 0; ; )
313 {
314 int j = k * 2 + 1;
315
316 if (j > Nm1)
317 break;
318
319 if (j < Nm1
320 && self->ranges [j]->next > self->ranges [j + 1]->next)
321 ++j;
322
323 if (self->ranges [j]->next >= elem->next)
324 break;
325
326 self->ranges [k] = self->ranges [j];
327
328 k = j;
329 }
330
331 self->ranges [k] = elem;
332}
333
334static void
335upheap (PINGER *self, int k)
336{
337 RANGE *elem = self->ranges [k];
338
339 while (k)
340 {
341 int j = (k - 1) >> 1;
342
343 if (self->ranges [j]->next <= elem->next)
344 break;
345
346 self->ranges [k] = self->ranges [j];
347
348 k = j;
349 }
350
351 self->ranges [k] = elem;
352}
353
143static void 354static void *
144inc_addr (addr_t *addr) 355ping_proc (void *self_)
145{ 356{
146 int len = sizeof (addr_t) - 1; 357 PINGER *self = (PINGER *)self_;
358 PKT pkt;
359
360 memset (&pkt, 0, sizeof (pkt));
361
362 tstamp now = NOW ();
363 tstamp next = now;
364
365 pkt.code = 0;
366 pkt.id = self->magic1;
367 pkt.seq = self->magic2;
368 pkt.magic = self->magic3;
369 pkt.pinger = self->id;
370
371 while (self->rangecnt)
372 {
373 RANGE *range = self->ranges [0];
374 int n, k;
375
376 // ranges [0] is always the next range to ping
377 tstamp wait = range->next - now;
378
379 // compare with the global frequency limit
380 {
381 tstamp diff = next - now;
382
383 if (wait < diff)
384 wait = diff;
385 else if (range)
386 next = range->next;
387 }
388
389 if (wait > 0.)
390 ssleep (wait);
391
392 now = NOW ();
393
394 ts_to_pkt (&pkt, now);
395
396 if (!range_send_ping (range, &pkt))
397 {
398 self->ranges [0] = self->ranges [--self->rangecnt];
399 range_free (range);
400 }
401 else
402 {
403 range->next = next;
404 range->next += range->interval;
405 }
406
407 downheap (self);
408
409 next += self->interval;
410 }
411
412 ssleep (self->maxrtt);
413
414 {
415 uint16_t id = self->id;
416
417 write (thr_res [1], &id, sizeof (id));
418 }
419
420 return 0;
421}
422
423/*****************************************************************************/
424
425static void
426pinger_start (PINGER *self)
427{
428 sigset_t fullsigset, oldsigset;
429 pthread_attr_t attr;
430
431 if (self->running)
432 return;
433
434 sigfillset (&fullsigset);
435
436 pthread_attr_init (&attr);
437 pthread_attr_setstacksize (&attr, PTHREAD_STACK_MIN < sizeof (long) * 2048 ? sizeof (long) * 2048 : PTHREAD_STACK_MIN);
438
439 pthread_sigmask (SIG_SETMASK, &fullsigset, &oldsigset);
440
441 if (pthread_create (&self->thrid, &attr, ping_proc, (void *)self))
442 croak ("AnyEvent::FastPing: unable to create pinger thread");
443
444 pthread_sigmask (SIG_SETMASK, &oldsigset, 0);
445
446 self->running = 1;
447}
448
449static void
450pinger_stop (PINGER *self)
451{
452 if (!self->running)
453 return;
454
455 self->running = 0;
456 pthread_cancel (self->thrid);
457 pthread_join (self->thrid, 0);
458}
459
460static void
461pinger_init (PINGER *self)
462{
463 memset (self, 0, sizeof (PINGER));
464
465 if (firstfree >= 0)
466 {
467 self->id = firstfree;
468 firstfree = pingerfree [firstfree];
469 }
470 else if (pingercnt == 0xffff)
471 croak ("unable to create more than 65536 AnyEvent::FastPing objects");
472 else
473 {
474 if (pingercnt == pingermax)
475 {
476 pingermax = pingermax * 2 + 16;
477 pingers = realloc (pingers , sizeof (pingers [0]) * pingermax);
478 pingerfree = realloc (pingerfree, sizeof (pingerfree [0]) * pingermax);
479 }
480
481 self->id = pingercnt++;
482 }
483
484 pingers [self->id] = self;
485
486 self->recvcb = &PL_sv_undef;
487 self->interval = MIN_INTERVAL;
488 self->maxrtt = 0.5;
489 self->rangemax = 16;
490 self->ranges = malloc (sizeof (self->ranges [0]) * self->rangemax);
491}
492
493static void
494pinger_free (PINGER *self)
495{
496 pinger_stop (self);
497
498 pingers [self->id] = 0;
499
500 SvREFCNT_dec (self->recvq);
501 SvREFCNT_dec (self->recvcb);
502
503 pingerfree [self->id] = firstfree;
504 firstfree = self->id;
505
506 while (self->rangecnt)
507 range_free (self->ranges [--self->rangecnt]);
508
509 free (self->ranges);
510}
511
512static void
513pinger_add_range (PINGER *self, RANGE *range)
514{
515 if (self->rangecnt == self->rangemax)
516 self->ranges = realloc (self->ranges, sizeof (self->ranges [0]) * (self->rangemax <<= 1));
517
518 self->ranges [self->rangecnt] = range;
519 upheap (self, self->rangecnt);
520 ++self->rangecnt;
521}
522
523/*****************************************************************************/
524
525static void
526recv_feed (PINGER *self, void *addr, int addrlen, tstamp rtt)
527{
528 if (!self->recvq)
529 {
530 /* first seen this round */
531 if (!SvOK (self->recvcb))
532 return;
533
534 self->recvq = newAV ();
535
536 self->nextrecv = firstrecv;
537 firstrecv = self->id;
538 }
539
540 {
541 AV *pkt = newAV ();
542
543 av_extend (pkt, 2-1);
544
545 AvARRAY (pkt)[0] = newSVpvn (addr, addrlen);
546 AvARRAY (pkt)[1] = newSVnv (rtt);
547 AvFILLp (pkt) = 2-1;
548
549 av_push (self->recvq, newRV_noinc ((SV *)pkt));
550 }
551}
552
553static void
554recv_flush (void)
555{
556 if (firstrecv < 0)
557 return;
558
559 ENTER;
560 SAVETMPS;
147 561
148 do 562 do
149 { 563 {
150 if ((*addr)[len] != 0xff) 564 dSP;
151 { 565 PINGER *self = pingers [firstrecv];
152 ++(*addr)[len]; 566 firstrecv = self->nextrecv;
153 break;
154 }
155 567
156 (*addr)[len] = 0; 568 self->nextrecv = -1;
157 }
158 while (len--);
159}
160 569
161static void * 570 PUSHMARK (SP);
162ping_proc (void *unused) 571 XPUSHs (sv_2mortal (newRV_noinc ((SV *)self->recvq)));
163{ 572 self->recvq = 0;
164 PKT pkt; 573 PUTBACK;
165 struct sockaddr_in sa4; 574 call_sv (self->recvcb, G_DISCARD | G_VOID);
166#if IPV6
167 struct sockaddr_in6 sa6;
168#endif
169
170 memset (&pkt, 0, sizeof (pkt));
171
172 memset (&sa4, 0, sizeof (sa4));
173 sa4.sin_family = AF_INET;
174 sa4.sin_port = 0;
175#if IPV6
176 memset (&sa6, 0, sizeof (sa6));
177 sa6.sin6_family = AF_INET6;
178 sa6.sin6_port = 0;
179#endif
180
181 for (;;)
182 {
183 REQ *req;
184 int len = read (thr_send [0], &req, sizeof (req));
185
186 if (!len)
187 pthread_exit (0);
188 else if (len != sizeof (req))
189 {
190 perror ("AnyEvent::FastPing: short reead or read error");
191 pthread_exit ((void *)-1);
192 }
193
194 //TODO: bind to source address
195
196 pkt.code = 0;
197 pkt.id = (uint16_t)magic;
198 pkt.seq = (uint16_t)~magic;
199 pkt.payload = req->payload;
200
201 tstamp now = NOW ();
202 tstamp next = now;
203
204 {
205 int r;
206 for (r = req->nranges; r--; )
207 inc_addr (&req->ranges [r].hi);
208 } 575 }
576 while (firstrecv >= 0);
209 577
210 while (req->nranges) 578 FREETMPS;
211 { 579 LEAVE;
212 RANGE *range = req->ranges;
213
214 if (!memcmp (&range->lo, &range->hi, sizeof (addr_t)))
215 req->ranges [0] = req->ranges [--req->nranges];
216 else
217 {
218 // ranges [0] is always the next range to ping
219 tstamp wait = range->next - now;
220
221 // compare with the global frequency limit
222 {
223 tstamp diff = next - now;
224
225 if (wait < diff)
226 wait = diff;
227 else if (range)
228 next = range->next;
229 }
230
231 if (wait > 0.)
232 {
233 struct timespec ts;
234
235 ts.tv_sec = wait;
236 ts.tv_nsec = (wait - ts.tv_sec) * 1000000000.;
237
238 nanosleep (&ts, 0);
239 }
240
241 now = NOW ();
242
243 pkt.stamp = now;
244 pkt.cksum = 0;
245
246 if (range->family == AF_INET)
247 {
248 pkt.type = ICMP4_ECHO;
249 pkt.cksum = icmp_cksum (&pkt, sizeof (pkt));
250
251 memcpy (&sa4.sin_addr,
252 sizeof (addr_t) - sizeof (sa4.sin_addr) + (char *)&range->lo,
253 sizeof (sa4.sin_addr));
254
255 if (sendto (icmp4_fd, &pkt, sizeof (pkt), 0, (struct sockaddr *)&sa4, sizeof (sa4)) > 0)
256 errno = 0;
257 }
258 else
259 {
260#if IPV6
261 pkt.type = ICMP6_ECHO;
262
263 memcpy (&sa6.sin6_addr,
264 sizeof (addr_t) - sizeof (sa6.sin6_addr) + (char *)&range->lo,
265 sizeof (sa6.sin6_addr));
266
267 if (sendto (icmp6_fd, &pkt, sizeof (pkt), 0, (struct sockaddr *)&sa6, sizeof (sa6)) > 0)
268 errno = 0;
269#endif
270 }
271
272 if (errno == ENOBUFS)
273 {
274 struct timespec ts;
275
276 ts.tv_sec = 0;
277 ts.tv_nsec = DRAIN_INTERVAL * 1000000000;
278
279 nanosleep (&ts, 0);
280 }
281 else
282 {
283 inc_addr (&range->lo);
284
285 range->next = next;
286 range->next += range->interval;
287 }
288
289 next += req->interval;
290 }
291
292 // make a downheap operation
293 int k = 0;
294 int n = 0;
295 for (;;)
296 {
297 ++n;
298 int j = k * 2 + 1;
299
300 if (j >= req->nranges)
301 break;
302 else if (j < req->nranges - 1)
303 if (req->ranges [j].next > req->ranges [j + 1].next)
304 ++j;
305
306 if (req->ranges [j].next >= req->ranges [k].next)
307 break;
308
309 RANGE temp = req->ranges [k];
310 req->ranges [k] = req->ranges [j];
311 req->ranges [j] = temp;
312
313 k = j;
314 }
315 }
316
317 write (thr_recv [1], &req, sizeof (req));
318 }
319
320 return 0;
321} 580}
322 581
582/*****************************************************************************/
583
584#if 0
323static void 585static void
324feed_reply (AV *res_av) 586feed_reply (AV *res_av)
325{ 587{
326 if (av_len (res_av) < 0)
327 return;
328
329 dSP; 588 dSP;
330 SV *res = sv_2mortal (newRV_inc ((SV *)res_av)); 589 SV *res = sv_2mortal (newRV_inc ((SV *)res_av));
331 int i; 590 int i;
591
592 if (av_len (res_av) < 0)
593 return;
332 594
333 ENTER; 595 ENTER;
334 SAVETMPS; 596 SAVETMPS;
335 597
336 for (i = av_len (cbs) + 1; i--; ) 598 for (i = av_len (cbs) + 1; i--; )
… …
344 } 606 }
345 607
346 FREETMPS; 608 FREETMPS;
347 LEAVE; 609 LEAVE;
348} 610}
611#endif
349 612
350static void 613static void
351boot () 614boot ()
352{ 615{
353 sigset_t fullsigset, oldsigset;
354 pthread_attr_t attr;
355
356 if (pipe (thr_send) < 0)
357 croak ("AnyEvent::FastPing: unable to create send pipe");
358
359 if (pipe (thr_recv) < 0) 616 if (pipe (thr_res) < 0)
360 croak ("AnyEvent::FastPing: unable to create receive pipe"); 617 croak ("AnyEvent::FastPing: unable to create receive pipe");
361 618
619 sv_setiv (get_sv ("AnyEvent::FastPing::THR_RES_FD", 1), thr_res [0]);
620
362 icmp4_fd = socket (AF_INET, SOCK_RAW, IPPROTO_ICMP); 621 icmp4_fd = socket (AF_INET, SOCK_RAW, IPPROTO_ICMP);
622 fcntl (icmp4_fd, F_SETFL, O_NONBLOCK);
363#ifdef ICMP_FILTER 623#ifdef ICMP_FILTER
364 { 624 {
365 struct icmp_filter oval; 625 struct icmp_filter oval;
366 oval.data = 0xffffffff & ~(1 << ICMP4_ECHO_REPLY); 626 oval.data = 0xffffffff & ~(1 << ICMP4_ECHO_REPLY);
367 setsockopt (icmp4_fd, SOL_RAW, ICMP_FILTER, &oval, sizeof oval); 627 setsockopt (icmp4_fd, SOL_RAW, ICMP_FILTER, &oval, sizeof oval);
368 } 628 }
369#endif 629#endif
370 630
371#if IPV6 631#if ENABLE_IPV6
372 icmp6_fd = socket (AF_INET6, SOCK_RAW, IPPROTO_ICMPV6); 632 icmp6_fd = socket (AF_INET6, SOCK_RAW, IPPROTO_ICMPV6);
633 fcntl (icmp6_fd, F_SETFL, O_NONBLOCK);
373# ifdef ICMP6_FILTER 634# ifdef ICMP6_FILTER
374 { 635 {
375 struct icmp6_filter oval; 636 struct icmp6_filter oval;
376 ICMP6_FILTER_SETBLOCKALL (&oval); 637 ICMP6_FILTER_SETBLOCKALL (&oval);
377 ICMP6_FILTER_SETPASS (ICMP6_ECHO_REPLY, &oval); 638 ICMP6_FILTER_SETPASS (ICMP6_ECHO_REPLY, &oval);
378 setsockopt (icmp6_fd, IPPROTO_ICMPV6, ICMP6_FILTER, &oval, sizeof oval); 639 setsockopt (icmp6_fd, IPPROTO_ICMPV6, ICMP6_FILTER, &oval, sizeof oval);
379 } 640 }
380# endif 641# endif
381#endif 642#endif
382 643
383 pthread_attr_init (&attr);
384 pthread_attr_setdetachstate (&attr, PTHREAD_CREATE_DETACHED);
385#ifdef PTHREAD_SCOPE_PROCESS
386 pthread_attr_setscope (&attr, PTHREAD_SCOPE_PROCESS);
387#endif
388
389 sigfillset (&fullsigset);
390
391 pthread_sigmask (SIG_SETMASK, &fullsigset, &oldsigset);
392
393 if (pthread_create (&pthrid, &attr, ping_proc, 0))
394 croak ("AnyEvent::FastPing: unable to create pinger thread");
395
396 pthread_sigmask (SIG_SETMASK, &oldsigset, 0);
397
398 sv_setiv (get_sv ("AnyEvent::FastPing::THR_REQ_FD", 1), thr_send [1]);
399 sv_setiv (get_sv ("AnyEvent::FastPing::THR_RES_FD", 1), thr_recv [0]);
400
401 sv_setiv (get_sv ("AnyEvent::FastPing::ICMP4_FD", 1), icmp4_fd); 644 sv_setiv (get_sv ("AnyEvent::FastPing::ICMP4_FD", 1), icmp4_fd);
402 sv_setiv (get_sv ("AnyEvent::FastPing::ICMP6_FD", 1), icmp6_fd); 645 sv_setiv (get_sv ("AnyEvent::FastPing::ICMP6_FD", 1), icmp6_fd);
403} 646}
404 647
405MODULE = AnyEvent::FastPing PACKAGE = AnyEvent::FastPing 648MODULE = AnyEvent::FastPing PACKAGE = AnyEvent::FastPing PREFIX = pinger_
406 649
407BOOT: 650BOOT:
408{ 651{
409 HV *stash = gv_stashpv ("AnyEvent::FastPing", 1); 652 HV *stash = gv_stashpv ("AnyEvent::FastPing", 1);
410 653
411 cbs = get_av ("AnyEvent::FastPing::CB", 1); 654 if (sizeof (PKT) & 3)
412 magic = getpid () ^ MAGIC; 655 croak ("size of PKT structure is not a multiple of 4");
413 656
414 boot (); 657 boot ();
415 658
416 newCONSTSUB (stash, "ipv4_supported", newSViv (icmp4_fd >= 0)); 659 newCONSTSUB (stash, "ipv4_supported", newSViv (icmp4_fd >= 0));
417 newCONSTSUB (stash, "ipv6_supported", newSViv (icmp6_fd >= 0)); 660 newCONSTSUB (stash, "ipv6_supported", newSViv (icmp6_fd >= 0));
… …
420 newCONSTSUB (stash, "icmp6_pktsize", newSViv (HDR_SIZE_IP6 + sizeof (PKT))); 663 newCONSTSUB (stash, "icmp6_pktsize", newSViv (HDR_SIZE_IP6 + sizeof (PKT)));
421} 664}
422 665
423PROTOTYPES: DISABLE 666PROTOTYPES: DISABLE
424 667
425SV *
426_req_icmp_ping (SV *ranges, NV interval, U32 payload, SV *id)
427 CODE:
428{
429 if (!SvROK (ranges) || SvTYPE (SvRV (ranges)) != SVt_PVAV)
430 croak ("address ranges must be given as arrayref with lo, hi pairs");
431
432 AV *rav = (AV *)SvRV (ranges);
433 int nranges = av_len (rav) + 1;
434
435 REQ *req = malloc (sizeof (REQ));
436 int i;
437
438 if (interval < MIN_INTERVAL)
439 interval = MIN_INTERVAL;
440
441 req->id = newSVsv (id);
442 req->interval = interval;
443 req->payload = payload;
444 req->nranges = nranges;
445 req->ranges = (RANGE *)malloc (nranges * sizeof (RANGE));
446
447 while (nranges--)
448 {
449 SV *sv = *av_fetch (rav, nranges, 1);
450
451 if (!SvROK (sv) || SvTYPE (SvRV (sv)) != SVt_PVAV)
452 croak ("address range must be given as arrayref with lo, hi, interval arrayrefs");
453
454 AV *av = (AV *)SvRV (sv);
455 RANGE *r = req->ranges + nranges;
456
457 SV *lo = *av_fetch (av, 0, 1);
458 SV *hi = *av_fetch (av, 1, 1);
459
460 sv_utf8_downgrade (lo, 0);
461 sv_utf8_downgrade (hi, 0);
462
463 memset (&r->lo, 0, sizeof (addr_t));
464 memset (&r->hi, 0, sizeof (addr_t));
465
466 if (SvPOKp (lo) && SvPOKp (hi))
467 {
468 if (SvCUR (lo) != SvCUR (hi))
469 croak ("addresses in range must be of the same size (either 4 or 16 bytes)");
470
471 if (SvCUR (lo) == 4)
472 {
473 r->family = AF_INET;
474 memcpy (sizeof (addr_t) - 4 + (char *)&r->lo, SvPVX (lo), 4);
475 memcpy (sizeof (addr_t) - 4 + (char *)&r->hi, SvPVX (hi), 4);
476 }
477 else if (SvCUR (lo) == 16)
478 {
479#if IPV6
480 r->family = AF_INET6;
481 memcpy (&r->lo, SvPVX (lo), sizeof (addr_t));
482 memcpy (&r->hi, SvPVX (hi), sizeof (addr_t));
483#else
484 croak ("IPv6 not supported in this configuration");
485#endif
486 }
487 else
488 croak ("addresses in range must be either 4 (IPv4) or 16 (IPV6) bytes in length");
489 }
490 else if (SvIOK (lo) && SvIOK (hi))
491 {
492 r->family = AF_INET;
493
494 uint32_t addr;
495 addr = htonl (SvUV (lo)); memcpy (sizeof (addr_t) - 4 + (char *)&r->lo, &addr, 4);
496 addr = htonl (SvUV (hi)); memcpy (sizeof (addr_t) - 4 + (char *)&r->hi, &addr, 4);
497 }
498 else
499 croak ("addresses in range must be strings with either 4 (IPv4) or 16 (IPv6) octets");
500
501 if (r->family == AF_INET)
502 {
503 if (icmp4_fd < 0)
504 croak ("AnyEvent::FastPing: IPv4 ping support not available on this system");
505 }
506 else
507 {
508 if (icmp6_fd < 0)
509 croak ("AnyEvent::FastPing: IPv6 ping support not available on this system");
510 }
511
512 r->interval = SvNV (*av_fetch (av, 2, 1));
513
514 if (r->interval < req->interval)
515 r->interval = req->interval;
516
517 r->next = 0.;
518 }
519
520 RETVAL = newSVpvn ((char *)&req, sizeof (req));
521}
522 OUTPUT:
523 RETVAL
524
525SV *
526_read_res ()
527 CODE:
528{
529 REQ *req;
530
531 if (read (thr_recv [0], &req, sizeof (req)) != sizeof (req))
532 RETVAL = &PL_sv_undef;
533
534 RETVAL = req->id;
535 free (req->ranges);
536 free (req);
537}
538 OUTPUT:
539 RETVAL
540
541void 668void
542_recv_icmp4 (...) 669_recv_icmp4 (...)
543 CODE: 670 CODE:
544{ 671{
545 char buf [512]; 672 char buf [512];
546 struct sockaddr_in sa; 673 struct sockaddr_in sa;
547 socklen_t sl = sizeof (sa); 674 int maxrecv;
548 AV *res_av = av_len (cbs) < 0 ? 0 : (AV *)sv_2mortal ((SV *)newAV ());
549 tstamp now = NOW ();
550 675
551 for (;;) 676 for (maxrecv = 256+1; --maxrecv; )
552 { 677 {
678 PINGER *pinger;
679 IP4HDR *iphdr = (IP4HDR *)buf;
680 socklen_t sl = sizeof (sa);
553 int len = recvfrom (icmp4_fd, buf, sizeof (buf), MSG_DONTWAIT | MSG_TRUNC, &sa, &sl); 681 int len = recvfrom (icmp4_fd, buf, sizeof (buf), MSG_TRUNC, (struct sockaddr *)&sa, &sl);
682 int hdrlen, totlen;
683 PKT *pkt;
554 684
555 if (len <= HDR_SIZE_IP4) 685 if (len <= HDR_SIZE_IP4)
556 break; 686 break;
557 687
558 IP4HDR *iphdr = (IP4HDR *)buf;
559
560 int hdrlen = (iphdr->version_ihl & 15) * 4; 688 hdrlen = (iphdr->version_ihl & 15) * 4;
561 int totlen = ntohs (iphdr->tot_len); 689 totlen = ntohs (iphdr->tot_len);
562 690
563 // packet corrupt?
564 if (!res_av
565 || totlen > len 691 if (totlen > len
566 || iphdr->protocol != IPPROTO_ICMP 692 || iphdr->protocol != IPPROTO_ICMP
567 || hdrlen < HDR_SIZE_IP4 || hdrlen + sizeof (PKT) != totlen) 693 || hdrlen < HDR_SIZE_IP4 || hdrlen + sizeof (PKT) != totlen)
568 continue; 694 continue;
569 695
570 PKT *pkt = (PKT *)(buf + hdrlen); 696 pkt = (PKT *)(buf + hdrlen);
571 697
572 if (pkt->type != ICMP4_ECHO_REPLY 698 if (pkt->type != ICMP4_ECHO_REPLY
573 || pkt->id != (uint16_t) magic 699 || pkt->pinger >= pingercnt
574 || pkt->seq != (uint16_t)~magic 700 || !pingers [pkt->pinger])
575 || !isnormal (pkt->stamp))
576 continue; 701 continue;
577 702
578 AV *av = newAV (); 703 pinger = pingers [pkt->pinger];
579 av_push (av, newSVpvn ((char *)&sa.sin_addr, 4));
580 av_push (av, newSVnv (now - pkt->stamp));
581 av_push (av, newSVuv (pkt->payload));
582 704
583 av_push (res_av, newRV_noinc ((SV *)av)); 705 if (!pkt_is_valid_for (pkt, pinger))
706 continue;
707
708 recv_feed (pinger, &sa.sin_addr, 4, NOW () - pkt_to_ts (pkt));
584 } 709 }
585 710
586 if (res_av) 711 recv_flush ();
587 feed_reply (res_av);
588} 712}
589 713
590void 714void
591_recv_icmp6 (...) 715_recv_icmp6 (...)
592 CODE: 716 CODE:
593{ 717{
594 struct sockaddr_in6 sa; 718 struct sockaddr_in6 sa;
595 socklen_t sl = sizeof (sa);
596 AV *res_av = av_len (cbs) < 0 ? 0 : (AV *)sv_2mortal ((SV *)newAV ());
597 PKT pkt; 719 PKT pkt;
598 tstamp now = NOW (); 720 int maxrecv;
599 721
600 for (;;) 722 for (maxrecv = 256+1; --maxrecv; )
601 { 723 {
724 PINGER *pinger;
725 socklen_t sl = sizeof (sa);
602 int len = recvfrom (icmp6_fd, &pkt, sizeof (pkt), MSG_DONTWAIT | MSG_TRUNC, &sa, &sl); 726 int len = recvfrom (icmp6_fd, &pkt, sizeof (pkt), MSG_TRUNC, (struct sockaddr *)&sa, &sl);
603 727
604 if (len != sizeof (PKT)) 728 if (len != sizeof (PKT))
605 break; 729 break;
606 730
607 if (!res_av
608 || pkt.type != ICMP6_ECHO_REPLY 731 if (pkt.type != ICMP6_ECHO_REPLY
609 || pkt.id != (uint16_t) magic 732 || pkt.pinger >= pingercnt
610 || pkt.seq != (uint16_t)~magic 733 || !pingers [pkt.pinger])
611 || !isnormal (pkt.stamp))
612 continue; 734 continue;
613 735
614 AV *av = newAV (); 736 pinger = pingers [pkt.pinger];
615 av_push (av, newSVpvn ((char *)&sa.sin6_addr, 16));
616 av_push (av, newSVnv (now - pkt.stamp));
617 av_push (av, newSVuv (pkt.payload));
618 737
619 av_push (res_av, newRV_noinc ((SV *)av)); 738 if (!pkt_is_valid_for (&pkt, pinger))
739 continue;
740
741 recv_feed (pinger, &sa.sin6_addr, 16, NOW () - pkt_to_ts (&pkt));
620 } 742 }
621 743
622 if (res_av) 744 recv_flush ();
623 feed_reply (res_av);
624} 745}
625 746
747void
748_new (SV *klass, UV magic1, UV magic2, UV magic3)
749 PPCODE:
750{
751 SV *pv = NEWSV (0, sizeof (PINGER));
752 PINGER *self = (PINGER *)SvPVX (pv);
753 SvPOK_only (pv);
754 XPUSHs (sv_2mortal (sv_bless (newRV_noinc (pv), gv_stashpv (SvPVutf8_nolen (klass), 1))));
755 pinger_init (self);
756 self->magic1 = magic1;
757 self->magic2 = magic2;
758 self->magic3 = magic3;
759}
760
761void
762_free (PINGER *self)
763 CODE:
764 pinger_free (self);
765
766IV
767id (PINGER *self, ...)
768 CODE:
769 RETVAL = self->id;
770 OUTPUT:
771 RETVAL
772
773void pinger_start (PINGER *self)
774
775void pinger_stop (PINGER *self)
776
777void
778_stop_id (UV id)
779 CODE:
780 if (id < pingercnt && pingers [id])
781 pinger_stop (pingers [id]);
782
783void
784interval (PINGER *self, NV interval)
785 CODE:
786 self->interval = interval > MIN_INTERVAL ? interval : MIN_INTERVAL;
787
788void
789max_rtt (PINGER *self, NV maxrtt)
790 CODE:
791 self->maxrtt = maxrtt;
792
793void
794add_range (PINGER *self, SV *lo_, SV *hi_, NV interval)
795 CODE:
796{
797 STRLEN lo_len, hi_len;
798 char *lo = SvPVbyte (lo_, lo_len);
799 char *hi = SvPVbyte (hi_, hi_len);
800 RANGE *range;
801
802 if (lo_len != hi_len || (lo_len != 4 && lo_len != 16))
803 croak ("AnyEvent::FastPing::add_range address range must be specified as two binary IPv4 or IPv6 addresses");
804
805 if (lo_len == 4 && icmp4_fd < 0) croak ("IPv4 support unavailable");
806 if (lo_len == 16 && icmp6_fd < 0) croak ("IPv6 support unavailable");
807
808 if (memcmp (lo, hi, lo_len) > 0)
809 croak ("AnyEvent::FastPing::add_range called with lo > hi");
810
811 range = calloc (1, sizeof (RANGE));
812
813 range->next = 0;
814 range->interval = interval > MIN_INTERVAL ? interval : MIN_INTERVAL;
815 range->addrlen = lo_len;
816
817 memcpy (sizeof (addr_tt) - lo_len + (char *)&range->lo, lo, lo_len);
818 memcpy (sizeof (addr_tt) - lo_len + (char *)&range->hi, hi, lo_len);
819
820 pinger_add_range (self, range);
821}
822
823void
824add_hosts (PINGER *self, SV *addrs, NV interval)
825 CODE:
826{
827 AV *av;
828 int i;
829 int cnt;
830 int addrlen;
831 RANGE *range;
832
833 if (!SvROK (addrs) || SvTYPE (SvRV (addrs)) != SVt_PVAV)
834 croak ("AnyEvent::FastPing::add_hosst expects an arrayref with binary IPv4 or IPv6 addresses");
835
836 av = (AV *)SvRV (addrs);
837 cnt = av_len (av) + 1;
838
839 if (!cnt)
840 XSRETURN_EMPTY;
841
842 addrlen = SvCUR (*av_fetch (av, 0, 1));
843
844 if (addrlen != 4 && addrlen != 16)
845 croak ("AnyEvent::FastPing::add_hosts addresses must be specified as binary IPv4 or IPv6 addresses");
846
847 for (i = cnt; --i; )
848 {
849 SV *sv = *av_fetch (av, i, 1);
850
851 if (!sv_utf8_downgrade (sv, 1) || addrlen != SvCUR (sv))
852 croak ("AnyEvent::FastPing::add_hosts addresses must all have the same size");
853 }
854
855 range = calloc (1, sizeof (RANGE) + cnt * addrlen);
856
857 range->next = 0;
858 range->interval = interval > MIN_INTERVAL ? interval : MIN_INTERVAL;
859 range->addrlen = addrlen;
860 range->addrcnt = cnt;
861
862 for (i = cnt; i--; )
863 memcpy ((uint8_t *)(range + 1) + (cnt - 1 - i) * addrlen,
864 SvPVbyte_nolen (*av_fetch (av, i, 1)),
865 addrlen);
866
867 pinger_add_range (self, range);
868}
869
870void
871on_recv (PINGER *self, SV *cb)
872 CODE:
873 SvREFCNT_dec (self->recvcb);
874 self->recvcb = newSVsv (cb);
875

Diff Legend

– Removed lines
+ Added lines
< Changed lines
> Changed lines