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

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

Diff Legend

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

Comparing AnyEvent-FastPing/FastPing.xs (file contents): Revision 1.5 by root, Fri Jul 3 09:26:19 2009 UTC vs. Revision 1.11 by root, Tue Feb 1 04:06:24 2011 UTC

Diff Legend

Comparing AnyEvent-FastPing/FastPing.xs (file contents):
Revision 1.5 by root, Fri Jul 3 09:26:19 2009 UTC vs.
Revision 1.11 by root, Tue Feb 1 04:06:24 2011 UTC