ViewVC Help
View File | Revision Log | Show Annotations | Download File
/cvs/AnyEvent-FastPing/FastPing.xs
Revision: 1.7
Committed: Sun Jan 16 17:05:32 2011 UTC (13 years, 4 months ago) by root
Branch: MAIN
Changes since 1.6: +1 -1 lines
Log Message:
*** empty log message ***

File Contents

# Content
1 #if defined(__linux)
2 # define ENABLE_IPV6 1 // if you get compilation problems try to disable IPv6
3 #else
4 # define ENABLE_IPV6 0
5 #endif
6
7 #include "EXTERN.h"
8 #include "perl.h"
9 #include "XSUB.h"
10
11 #include <pthread.h>
12
13 #include <math.h>
14 #include <stdio.h>
15 #include <stdlib.h>
16 #include <string.h>
17
18 #include <time.h>
19 #include <poll.h>
20 #include <unistd.h>
21 #include <inttypes.h>
22 #include <fcntl.h>
23 #include <errno.h>
24
25 #include <sys/types.h>
26 #include <sys/time.h>
27 #include <sys/socket.h>
28
29 #include <netinet/in.h>
30 #include <arpa/inet.h>
31
32 #ifdef __linux
33 # include <linux/icmp.h>
34 #endif
35 #if ENABLE_IPV6
36 # include <netinet/icmp6.h>
37 #endif
38
39 #define ICMP4_ECHO 8
40 #define ICMP4_ECHO_REPLY 0
41 #define ICMP6_ECHO 128
42 #define ICMP6_ECHO_REPLY 129
43
44 #define DRAIN_INTERVAL .000001 // how long to wait when sendto returns ENOBUFS, in seconds
45 #define MIN_INTERVAL .000001 // minimum packet send interval, in seconds
46
47 #define HDR_SIZE_IP4 20
48 #define HDR_SIZE_IP6 48
49
50 //TODO: xread/xwrite for atomicity? we currently rely on the fact that the pip buffersize divides exactly by pointer sizes
51
52 typedef uint8_t addr_t[16];
53
54 typedef double tstamp;
55
56 static tstamp
57 NOW (void)
58 {
59 struct timeval tv;
60 gettimeofday (&tv, 0);
61 return tv.tv_sec + tv.tv_usec * 0.000001;
62 }
63
64 typedef struct {
65 int family;
66 addr_t lo, hi;
67 double interval;
68 tstamp next;
69 } RANGE;
70
71 typedef struct {
72 SV *id;
73 double interval;
74 int nranges;
75 RANGE *ranges;
76 uint32_t payload;
77 } REQ;
78
79 typedef struct {
80 uint8_t version_ihl;
81 uint8_t tos;
82 uint16_t tot_len;
83
84 uint16_t id;
85 uint16_t flags;
86
87 uint8_t ttl;
88 uint8_t protocol;
89 uint16_t cksum;
90
91 uint32_t src;
92 uint32_t dst;
93 } IP4HDR;
94
95 typedef struct {
96 uint8_t version;
97 uint8_t x1, x2, x3;
98
99 uint16_t payload_len;
100 uint8_t nxt_hdr;
101 uint8_t hop_limit;
102
103 uint8_t src[16];
104 uint8_t dst[16];
105 } IP6HDR;
106
107 #define MAGIC 0xca4c
108
109 static uint16_t magic;
110
111 typedef struct {
112 uint8_t type, code;
113 uint16_t cksum;
114 uint16_t id, seq;
115 uint32_t payload;
116 tstamp stamp; // be careful when accessing this
117 } PKT;
118
119 static pthread_t pthrid;
120 static int thr_send[2]; // send to worker
121 static int thr_recv[2]; // receive from worker
122
123 static int icmp4_fd, icmp6_fd;
124
125 static AV *cbs;
126
127 static uint16_t
128 icmp_cksum (void *data, unsigned int len)
129 {
130 register int sum = 0;
131 uint16_t *wp;
132
133 assert (~len & 1);
134
135 for (wp = (uint16_t *)data; len; wp++, len -= 2)
136 sum += *wp;
137
138 sum = (sum >> 16) + (sum & 0xffff); /* add high 16 to low 16 */
139 sum += sum >> 16; /* add carry */
140
141 return ~sum;
142 }
143
144 static void
145 inc_addr (addr_t *addr)
146 {
147 int len = sizeof (addr_t) - 1;
148
149 do
150 {
151 if ((*addr)[len] != 0xff)
152 {
153 ++(*addr)[len];
154 break;
155 }
156
157 (*addr)[len] = 0;
158 }
159 while (len--);
160 }
161
162 static void *
163 ping_proc (void *unused)
164 {
165 PKT pkt;
166 struct sockaddr_in sa4;
167 #if ENABLE_IPV6
168 struct sockaddr_in6 sa6;
169 #endif
170
171 memset (&pkt, 0, sizeof (pkt));
172
173 memset (&sa4, 0, sizeof (sa4));
174 sa4.sin_family = AF_INET;
175 sa4.sin_port = 0;
176 #if ENABLE_IPV6
177 memset (&sa6, 0, sizeof (sa6));
178 sa6.sin6_family = AF_INET6;
179 sa6.sin6_port = 0;
180 #endif
181
182 for (;;)
183 {
184 REQ *req;
185 int len = read (thr_send [0], &req, sizeof (req));
186
187 tstamp now = NOW ();
188 tstamp next = now;
189
190 if (!len)
191 pthread_exit (0);
192 else if (len != sizeof (req))
193 {
194 perror ("AnyEvent::FastPing: short read or read error");
195 pthread_exit ((void *)-1);
196 }
197
198 //TODO: bind to source address
199
200 pkt.code = 0;
201 pkt.id = (uint16_t)magic;
202 pkt.seq = (uint16_t)~magic;
203 pkt.payload = req->payload;
204
205 {
206 int r;
207 for (r = req->nranges; r--; )
208 inc_addr (&req->ranges [r].hi);
209 }
210
211 while (req->nranges)
212 {
213 RANGE *range = req->ranges;
214 int n, k;
215
216 if (!memcmp (&range->lo, &range->hi, sizeof (addr_t)))
217 req->ranges [0] = req->ranges [--req->nranges];
218 else
219 {
220 // ranges [0] is always the next range to ping
221 tstamp wait = range->next - now;
222
223 // compare with the global frequency limit
224 {
225 tstamp diff = next - now;
226
227 if (wait < diff)
228 wait = diff;
229 else if (range)
230 next = range->next;
231 }
232
233 if (wait > 0.)
234 {
235 struct timespec ts;
236
237 ts.tv_sec = wait;
238 ts.tv_nsec = (wait - ts.tv_sec) * 1000000000.;
239
240 nanosleep (&ts, 0);
241 }
242
243 now = NOW ();
244
245 pkt.stamp = now;
246 pkt.cksum = 0;
247
248 if (range->family == AF_INET)
249 {
250 pkt.type = ICMP4_ECHO;
251 pkt.cksum = icmp_cksum (&pkt, sizeof (pkt));
252
253 memcpy (&sa4.sin_addr,
254 sizeof (addr_t) - sizeof (sa4.sin_addr) + (char *)&range->lo,
255 sizeof (sa4.sin_addr));
256
257 if (sendto (icmp4_fd, &pkt, sizeof (pkt), 0, (struct sockaddr *)&sa4, sizeof (sa4)) > 0)
258 errno = 0;
259 }
260 else
261 {
262 #if ENABLE_IPV6
263 pkt.type = ICMP6_ECHO;
264
265 memcpy (&sa6.sin6_addr,
266 sizeof (addr_t) - sizeof (sa6.sin6_addr) + (char *)&range->lo,
267 sizeof (sa6.sin6_addr));
268
269 if (sendto (icmp6_fd, &pkt, sizeof (pkt), 0, (struct sockaddr *)&sa6, sizeof (sa6)) > 0)
270 errno = 0;
271 #endif
272 }
273
274 if (errno == ENOBUFS)
275 {
276 struct timespec ts;
277
278 ts.tv_sec = 0;
279 ts.tv_nsec = DRAIN_INTERVAL * 1000000000;
280
281 nanosleep (&ts, 0);
282 }
283 else
284 {
285 inc_addr (&range->lo);
286
287 range->next = next;
288 range->next += range->interval;
289 }
290
291 next += req->interval;
292 }
293
294 // make a downheap operation
295 for (n = k = 0; ; )
296 {
297 int j = k * 2 + 1;
298
299 ++n;
300
301 if (j >= req->nranges)
302 break;
303 else if (j < req->nranges - 1)
304 if (req->ranges [j].next > req->ranges [j + 1].next)
305 ++j;
306
307 if (req->ranges [j].next >= req->ranges [k].next)
308 break;
309
310 {
311 RANGE temp = req->ranges [k];
312 req->ranges [k] = req->ranges [j];
313 req->ranges [j] = temp;
314 }
315
316 k = j;
317 }
318 }
319
320 write (thr_recv [1], &req, sizeof (req));
321 }
322
323 return 0;
324 }
325
326 static void
327 feed_reply (AV *res_av)
328 {
329 dSP;
330 SV *res = sv_2mortal (newRV_inc ((SV *)res_av));
331 int i;
332
333 if (av_len (res_av) < 0)
334 return;
335
336 ENTER;
337 SAVETMPS;
338
339 for (i = av_len (cbs) + 1; i--; )
340 {
341 SV *cb = *av_fetch (cbs, i, 1);
342
343 PUSHMARK (SP);
344 XPUSHs (res);
345 PUTBACK;
346 call_sv (cb, G_DISCARD | G_VOID);
347 }
348
349 FREETMPS;
350 LEAVE;
351 }
352
353 static void
354 boot ()
355 {
356 sigset_t fullsigset, oldsigset;
357 pthread_attr_t attr;
358
359 if (pipe (thr_send) < 0)
360 croak ("AnyEvent::FastPing: unable to create send pipe");
361
362 if (pipe (thr_recv) < 0)
363 croak ("AnyEvent::FastPing: unable to create receive pipe");
364
365 icmp4_fd = socket (AF_INET, SOCK_RAW, IPPROTO_ICMP);
366 fcntl (icmp4_fd, F_SETFL, O_NONBLOCK);
367 #ifdef ICMP_FILTER
368 {
369 struct icmp_filter oval;
370 oval.data = 0xffffffff & ~(1 << ICMP4_ECHO_REPLY);
371 setsockopt (icmp4_fd, SOL_RAW, ICMP_FILTER, &oval, sizeof oval);
372 }
373 #endif
374
375 #if ENABLE_IPV6
376 icmp6_fd = socket (AF_INET6, SOCK_RAW, IPPROTO_ICMPV6);
377 fcntl (icmp6_fd, F_SETFL, O_NONBLOCK);
378 # ifdef ICMP6_FILTER
379 {
380 struct icmp6_filter oval;
381 ICMP6_FILTER_SETBLOCKALL (&oval);
382 ICMP6_FILTER_SETPASS (ICMP6_ECHO_REPLY, &oval);
383 setsockopt (icmp6_fd, IPPROTO_ICMPV6, ICMP6_FILTER, &oval, sizeof oval);
384 }
385 # endif
386 #endif
387
388 pthread_attr_init (&attr);
389 pthread_attr_setdetachstate (&attr, PTHREAD_CREATE_DETACHED);
390 #ifdef PTHREAD_SCOPE_PROCESS
391 pthread_attr_setscope (&attr, PTHREAD_SCOPE_PROCESS);
392 #endif
393
394 sigfillset (&fullsigset);
395
396 pthread_sigmask (SIG_SETMASK, &fullsigset, &oldsigset);
397
398 if (pthread_create (&pthrid, &attr, ping_proc, 0))
399 croak ("AnyEvent::FastPing: unable to create pinger thread");
400
401 pthread_sigmask (SIG_SETMASK, &oldsigset, 0);
402
403 sv_setiv (get_sv ("AnyEvent::FastPing::THR_REQ_FD", 1), thr_send [1]);
404 sv_setiv (get_sv ("AnyEvent::FastPing::THR_RES_FD", 1), thr_recv [0]);
405
406 sv_setiv (get_sv ("AnyEvent::FastPing::ICMP4_FD", 1), icmp4_fd);
407 sv_setiv (get_sv ("AnyEvent::FastPing::ICMP6_FD", 1), icmp6_fd);
408 }
409
410 MODULE = AnyEvent::FastPing PACKAGE = AnyEvent::FastPing
411
412 BOOT:
413 {
414 HV *stash = gv_stashpv ("AnyEvent::FastPing", 1);
415
416 cbs = get_av ("AnyEvent::FastPing::CB", 1);
417 magic = getpid () ^ MAGIC;
418
419 boot ();
420
421 newCONSTSUB (stash, "ipv4_supported", newSViv (icmp4_fd >= 0));
422 newCONSTSUB (stash, "ipv6_supported", newSViv (icmp6_fd >= 0));
423
424 newCONSTSUB (stash, "icmp4_pktsize", newSViv (HDR_SIZE_IP4 + sizeof (PKT)));
425 newCONSTSUB (stash, "icmp6_pktsize", newSViv (HDR_SIZE_IP6 + sizeof (PKT)));
426 }
427
428 PROTOTYPES: DISABLE
429
430 SV *
431 _req_icmp_ping (SV *ranges, NV interval, U32 payload, SV *id)
432 CODE:
433 {
434 AV *rav;
435 int nranges, i;
436 REQ *req;
437
438 if (!SvROK (ranges) || SvTYPE (SvRV (ranges)) != SVt_PVAV)
439 croak ("address ranges must be given as arrayref with lo, hi pairs");
440
441 rav = (AV *)SvRV (ranges);
442 nranges = av_len (rav) + 1;
443
444 req = malloc (sizeof (REQ));
445
446 if (interval < MIN_INTERVAL)
447 interval = MIN_INTERVAL;
448
449 req->id = newSVsv (id);
450 req->interval = interval;
451 req->payload = payload;
452 req->nranges = nranges;
453 req->ranges = (RANGE *)malloc (nranges * sizeof (RANGE));
454
455 while (nranges--)
456 {
457 SV *sv = *av_fetch (rav, nranges, 1);
458 SV *lo, *hi;
459 AV *av;
460 RANGE *r;
461
462 if (!SvROK (sv) || SvTYPE (SvRV (sv)) != SVt_PVAV)
463 croak ("address range must be given as arrayref with lo, hi, interval arrayrefs");
464
465 av = (AV *)SvRV (sv);
466 r = req->ranges + nranges;
467
468 lo = *av_fetch (av, 0, 1);
469 hi = *av_fetch (av, 1, 1);
470
471 sv_utf8_downgrade (lo, 0);
472 sv_utf8_downgrade (hi, 0);
473
474 memset (&r->lo, 0, sizeof (addr_t));
475 memset (&r->hi, 0, sizeof (addr_t));
476
477 if (SvPOKp (lo) && SvPOKp (hi))
478 {
479 if (SvCUR (lo) != SvCUR (hi))
480 croak ("all addresses in range must be of the same size (either 4 or 16 bytes)");
481
482 if (SvCUR (lo) == 4)
483 {
484 r->family = AF_INET;
485 memcpy (sizeof (addr_t) - 4 + (char *)&r->lo, SvPVX (lo), 4);
486 memcpy (sizeof (addr_t) - 4 + (char *)&r->hi, SvPVX (hi), 4);
487 }
488 else if (SvCUR (lo) == 16)
489 {
490 #if ENABLE_IPV6
491 r->family = AF_INET6;
492 memcpy (&r->lo, SvPVX (lo), sizeof (addr_t));
493 memcpy (&r->hi, SvPVX (hi), sizeof (addr_t));
494 #else
495 croak ("IPv6 not supported in this configuration");
496 #endif
497 }
498 else
499 croak ("addresses in range must be either 4 (IPv4) or 16 (IPv6) bytes in length");
500 }
501 else if (SvIOK (lo) && SvIOK (hi))
502 {
503 uint32_t addr;
504
505 r->family = AF_INET;
506
507 addr = htonl (SvUV (lo)); memcpy (sizeof (addr_t) - 4 + (char *)&r->lo, &addr, 4);
508 addr = htonl (SvUV (hi)); memcpy (sizeof (addr_t) - 4 + (char *)&r->hi, &addr, 4);
509 }
510 else
511 croak ("addresses in range must be strings with either 4 (IPv4) or 16 (IPv6) octets");
512
513 if (r->family == AF_INET)
514 {
515 if (icmp4_fd < 0)
516 croak ("AnyEvent::FastPing: IPv4 ping support not available on this system");
517 }
518 else
519 {
520 if (icmp6_fd < 0)
521 croak ("AnyEvent::FastPing: IPv6 ping support not available on this system");
522 }
523
524 r->interval = SvNV (*av_fetch (av, 2, 1));
525
526 if (r->interval < req->interval)
527 r->interval = req->interval;
528
529 r->next = 0.;
530 }
531
532 RETVAL = newSVpvn ((char *)&req, sizeof (req));
533 }
534 OUTPUT:
535 RETVAL
536
537 SV *
538 _read_res ()
539 CODE:
540 {
541 REQ *req;
542
543 if (read (thr_recv [0], &req, sizeof (req)) != sizeof (req))
544 RETVAL = &PL_sv_undef;
545
546 RETVAL = req->id;
547 free (req->ranges);
548 free (req);
549 }
550 OUTPUT:
551 RETVAL
552
553 void
554 _recv_icmp4 (...)
555 CODE:
556 {
557 char buf [512];
558 struct sockaddr_in sa;
559 socklen_t sl = sizeof (sa);
560 AV *res_av = av_len (cbs) < 0 ? 0 : (AV *)sv_2mortal ((SV *)newAV ());
561 tstamp now = NOW ();
562
563 for (;;)
564 {
565 IP4HDR *iphdr = (IP4HDR *)buf;
566 int len = recvfrom (icmp4_fd, buf, sizeof (buf), MSG_TRUNC, (struct sockaddr *)&sa, &sl);
567 int hdrlen, totlen;
568 PKT *pkt;
569
570 if (len <= HDR_SIZE_IP4)
571 break;
572
573 hdrlen = (iphdr->version_ihl & 15) * 4;
574 totlen = ntohs (iphdr->tot_len);
575
576 // packet corrupt?
577 if (!res_av
578 || totlen > len
579 || iphdr->protocol != IPPROTO_ICMP
580 || hdrlen < HDR_SIZE_IP4 || hdrlen + sizeof (PKT) != totlen)
581 continue;
582
583 pkt = (PKT *)(buf + hdrlen);
584
585 if (pkt->type != ICMP4_ECHO_REPLY
586 || pkt->id != (uint16_t) magic
587 || pkt->seq != (uint16_t)~magic
588 || !isnormal (pkt->stamp))
589 continue;
590
591 {
592 AV *av = newAV ();
593 av_push (av, newSVpvn ((char *)&sa.sin_addr, 4));
594 av_push (av, newSVnv (now - pkt->stamp));
595 av_push (av, newSVuv (pkt->payload));
596
597 av_push (res_av, newRV_noinc ((SV *)av));
598 }
599 }
600
601 if (res_av)
602 feed_reply (res_av);
603 }
604
605 void
606 _recv_icmp6 (...)
607 CODE:
608 {
609 struct sockaddr_in6 sa;
610 socklen_t sl = sizeof (sa);
611 AV *res_av = av_len (cbs) < 0 ? 0 : (AV *)sv_2mortal ((SV *)newAV ());
612 PKT pkt;
613 tstamp now = NOW ();
614
615 for (;;)
616 {
617 int len = recvfrom (icmp6_fd, &pkt, sizeof (pkt), MSG_TRUNC, (struct sockaddr *)&sa, &sl);
618
619 if (len != sizeof (PKT))
620 break;
621
622 if (!res_av
623 || pkt.type != ICMP6_ECHO_REPLY
624 || pkt.id != (uint16_t) magic
625 || pkt.seq != (uint16_t)~magic
626 || !isnormal (pkt.stamp))
627 continue;
628
629 {
630 AV *av = newAV ();
631 av_push (av, newSVpvn ((char *)&sa.sin6_addr, 16));
632 av_push (av, newSVnv (now - pkt.stamp));
633 av_push (av, newSVuv (pkt.payload));
634
635 av_push (res_av, newRV_noinc ((SV *)av));
636 }
637 }
638
639 if (res_av)
640 feed_reply (res_av);
641 }
642