ViewVC Help
View File | Revision Log | Show Annotations | Download File
/cvs/AnyEvent-FastPing/FPing.xs
Revision: 1.5
Committed: Fri May 4 15:33:56 2007 UTC (17 years ago) by root
Branch: MAIN
Changes since 1.4: +21 -4 lines
Log Message:
*** empty log message ***

File Contents

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