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

Comparing AnyEvent-FastPing/FastPing.xs (file contents):
Revision 1.4 by root, Sun Mar 29 20:07:03 2009 UTC vs.
Revision 1.14 by root, Wed Oct 8 21:12:28 2014 UTC

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

Diff Legend

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

Comparing AnyEvent-FastPing/FastPing.xs (file contents): Revision 1.4 by root, Sun Mar 29 20:07:03 2009 UTC vs. Revision 1.14 by root, Wed Oct 8 21:12:28 2014 UTC

Diff Legend

Comparing AnyEvent-FastPing/FastPing.xs (file contents):
Revision 1.4 by root, Sun Mar 29 20:07:03 2009 UTC vs.
Revision 1.14 by root, Wed Oct 8 21:12:28 2014 UTC