[ViewVC] Diff of: cvs/AnyEvent-FastPing/FastPing.xs

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

Diff Legend

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

Comparing AnyEvent-FastPing/FastPing.xs (file contents): Revision 1.1 by root, Sun Apr 27 15:43:51 2008 UTC vs. Revision 1.10 by root, Mon Jan 31 05:35:48 2011 UTC

Diff Legend

Comparing AnyEvent-FastPing/FastPing.xs (file contents):
Revision 1.1 by root, Sun Apr 27 15:43:51 2008 UTC vs.
Revision 1.10 by root, Mon Jan 31 05:35:48 2011 UTC