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

Comparing AnyEvent-FastPing/FastPing.xs (file contents):
Revision 1.6 by root, Sat Jul 18 00:53:42 2009 UTC vs.
Revision 1.12 by root, Wed Feb 2 19:26:45 2011 UTC

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

Diff Legend

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

Comparing AnyEvent-FastPing/FastPing.xs (file contents): Revision 1.6 by root, Sat Jul 18 00:53:42 2009 UTC vs. Revision 1.12 by root, Wed Feb 2 19:26:45 2011 UTC

Diff Legend

Comparing AnyEvent-FastPing/FastPing.xs (file contents):
Revision 1.6 by root, Sat Jul 18 00:53:42 2009 UTC vs.
Revision 1.12 by root, Wed Feb 2 19:26:45 2011 UTC