AnyEvent-FastPing/FastPing.xs

#if defined(__linux)
# define ENABLE_IPV6 1 // if you get compilation problems try to disable IPv6
#else
# define ENABLE_IPV6 0
#endif

#include "EXTERN.h"
#include "perl.h"
#include "XSUB.h"

#include <pthread.h>

#include <math.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>

#include <time.h>
#include <poll.h>
#include <unistd.h>
#include <inttypes.h>
#include <fcntl.h>
#include <errno.h>

#include <sys/types.h>
#include <sys/time.h>
#include <sys/socket.h>

#include <netinet/in.h>
#include <arpa/inet.h>

#ifdef __linux
# include <linux/icmp.h>
#endif
#if ENABLE_IPV6
# include <netinet/icmp6.h>
#endif

#define ICMP4_ECHO       8
#define ICMP4_ECHO_REPLY 0
#define ICMP6_ECHO       128
#define ICMP6_ECHO_REPLY 129

#define DRAIN_INTERVAL .000001 // how long to wait when sendto returns ENOBUFS, in seconds
#define MIN_INTERVAL   .000001 // minimum packet send interval, in seconds

#define HDR_SIZE_IP4  20
#define HDR_SIZE_IP6  48

//TODO: xread/xwrite for atomicity? we currently rely on the fact that the pip buffersize divides exactly by pointer sizes

typedef uint8_t addr_t[16];

typedef double tstamp;

static tstamp
NOW (void)
{
  struct timeval tv;
  gettimeofday (&tv, 0);
  return tv.tv_sec + tv.tv_usec * 0.000001;
}

typedef struct {
  int family;
  addr_t lo, hi;
  double interval;
  tstamp next;
} RANGE;

typedef struct {
  SV *id;
  double interval;
  int nranges;
  RANGE *ranges;
  uint32_t payload;
} REQ;

typedef struct {
  uint8_t version_ihl;
  uint8_t tos;
  uint16_t tot_len;

  uint16_t id;
  uint16_t flags;

  uint8_t ttl;
  uint8_t protocol;
  uint16_t cksum;

  uint32_t src;
  uint32_t dst;
} IP4HDR;

typedef struct {
  uint8_t version;
  uint8_t x1, x2, x3;

  uint16_t payload_len;
  uint8_t nxt_hdr;
  uint8_t hop_limit;

  uint8_t src[16];
  uint8_t dst[16];
} IP6HDR;

#define MAGIC 0xca4c

static uint16_t magic;

typedef struct {
  uint8_t type, code;
  uint16_t cksum;
  uint16_t id, seq;
  uint32_t payload;
  tstamp stamp; // be careful when accessing this
} PKT;

static pthread_t pthrid;
static int thr_send[2]; // send to worker
static int thr_recv[2]; // receive from worker

static int icmp4_fd, icmp6_fd;

static AV *cbs;

static uint16_t
icmp_cksum (void *data, unsigned int len)
{
  register int sum = 0;
  uint16_t *wp;

  assert (~len & 1);

  for (wp = (uint16_t *)data; len; wp++, len -= 2)
    sum += *wp;

  sum = (sum >> 16) + (sum & 0xffff);   /* add high 16 to low 16 */
  sum += sum >> 16;                     /* add carry */

  return ~sum;
}

static void
inc_addr (addr_t *addr)
{
  int len = sizeof (addr_t) - 1;

  do
    {
      if ((*addr)[len] != 0xff)
        {
          ++(*addr)[len];
          break;
        }

      (*addr)[len] = 0;
    }
  while (len--);
}

static void *
ping_proc (void *unused)
{
  PKT pkt;
  struct sockaddr_in sa4;
#if ENABLE_IPV6
  struct sockaddr_in6 sa6;
#endif

  memset (&pkt, 0, sizeof (pkt));

  memset (&sa4, 0, sizeof (sa4));
  sa4.sin_family  = AF_INET;
  sa4.sin_port    = 0;
#if ENABLE_IPV6
  memset (&sa6, 0, sizeof (sa6));
  sa6.sin6_family = AF_INET6;
  sa6.sin6_port   = 0;
#endif

  for (;;)
    {
      REQ *req;
      int len = read (thr_send [0], &req, sizeof (req));

      tstamp now = NOW ();
      tstamp next = now;

      if (!len)
        pthread_exit (0);
      else if (len != sizeof (req))
        {
          perror ("AnyEvent::FastPing: short reead or read error");
          pthread_exit ((void *)-1);
        }

      //TODO: bind to source address

      pkt.code    = 0;
      pkt.id      = (uint16_t)magic;
      pkt.seq     = (uint16_t)~magic;
      pkt.payload = req->payload;

      {
        int r;
        for (r = req->nranges; r--; )
          inc_addr (&req->ranges [r].hi);
      }

      while (req->nranges)
        {
          RANGE *range = req->ranges;
          int n, k;

          if (!memcmp (&range->lo, &range->hi, sizeof (addr_t)))
            req->ranges [0] = req->ranges [--req->nranges];
          else
            {
              // ranges [0] is always the next range to ping
              tstamp wait = range->next - now;

              // compare with the global frequency limit
              {
                tstamp diff = next - now;

                if (wait < diff)
                  wait = diff;
                else if (range)
                  next = range->next;
              }

              if (wait > 0.)
                {
                  struct timespec ts;

                  ts.tv_sec  = wait;
                  ts.tv_nsec = (wait - ts.tv_sec) * 1000000000.;

                  nanosleep (&ts, 0);
                }

              now = NOW ();

              pkt.stamp = now;
              pkt.cksum = 0;

              if (range->family == AF_INET)
                {
                  pkt.type  = ICMP4_ECHO;
                  pkt.cksum = icmp_cksum (&pkt, sizeof (pkt));

                  memcpy (&sa4.sin_addr,
                          sizeof (addr_t) - sizeof (sa4.sin_addr) + (char *)&range->lo,
                          sizeof (sa4.sin_addr));

                  if (sendto (icmp4_fd, &pkt, sizeof (pkt), 0, (struct sockaddr *)&sa4, sizeof (sa4)) > 0)
                    errno = 0;
                }
              else
                {
#if ENABLE_IPV6
                  pkt.type = ICMP6_ECHO;

                  memcpy (&sa6.sin6_addr,
                          sizeof (addr_t) - sizeof (sa6.sin6_addr) + (char *)&range->lo,
                          sizeof (sa6.sin6_addr));

                  if (sendto (icmp6_fd, &pkt, sizeof (pkt), 0, (struct sockaddr *)&sa6, sizeof (sa6)) > 0)
                    errno = 0;
#endif
                }

              if (errno == ENOBUFS)
                {
                  struct timespec ts;

                  ts.tv_sec  = 0;
                  ts.tv_nsec = DRAIN_INTERVAL * 1000000000;

                  nanosleep (&ts, 0);
                }
              else
                {
                  inc_addr (&range->lo);

                  range->next = next;
                  range->next += range->interval;
                }

              next += req->interval;
            }

          // make a downheap operation
          for (n = k = 0; ; )
            {
              int j = k * 2 + 1;

              ++n;

              if (j >= req->nranges)
                break;
              else if (j < req->nranges - 1)
                if (req->ranges [j].next > req->ranges [j + 1].next)
                  ++j;

              if (req->ranges [j].next >= req->ranges [k].next)
                break;

              {
                RANGE temp = req->ranges [k];
                req->ranges [k] = req->ranges [j];
                req->ranges [j] = temp;
              }

              k = j;
            }
        }

      write (thr_recv [1], &req, sizeof (req));
    }

  return 0;
}

static void
feed_reply (AV *res_av)
{
  dSP;
  SV *res = sv_2mortal (newRV_inc ((SV *)res_av));
  int i;

  if (av_len (res_av) < 0)
    return;

  ENTER;
  SAVETMPS;

  for (i = av_len (cbs) + 1; i--; )
    {
      SV *cb = *av_fetch (cbs, i, 1);

      PUSHMARK (SP);
      XPUSHs (res);
      PUTBACK;
      call_sv (cb, G_DISCARD | G_VOID);
    }

  FREETMPS;
  LEAVE;
}

static void
boot ()
{
  sigset_t fullsigset, oldsigset;
  pthread_attr_t attr;

  if (pipe (thr_send) < 0)
    croak ("AnyEvent::FastPing: unable to create send pipe");

  if (pipe (thr_recv) < 0)
    croak ("AnyEvent::FastPing: unable to create receive pipe");

  icmp4_fd = socket (AF_INET, SOCK_RAW, IPPROTO_ICMP);
  fcntl (icmp4_fd, F_SETFL, O_NONBLOCK);
#ifdef ICMP_FILTER
  {
    struct icmp_filter oval;
    oval.data = 0xffffffff & ~(1 << ICMP4_ECHO_REPLY);
    setsockopt (icmp4_fd, SOL_RAW, ICMP_FILTER, &oval, sizeof oval);
  }
#endif

#if ENABLE_IPV6
  icmp6_fd = socket (AF_INET6, SOCK_RAW, IPPROTO_ICMPV6);
  fcntl (icmp6_fd, F_SETFL, O_NONBLOCK);
# ifdef ICMP6_FILTER
  {
    struct icmp6_filter oval;
    ICMP6_FILTER_SETBLOCKALL (&oval);
    ICMP6_FILTER_SETPASS (ICMP6_ECHO_REPLY, &oval);
    setsockopt (icmp6_fd, IPPROTO_ICMPV6, ICMP6_FILTER, &oval, sizeof oval);
  }
# endif
#endif

  pthread_attr_init (&attr);
  pthread_attr_setdetachstate (&attr, PTHREAD_CREATE_DETACHED);
#ifdef PTHREAD_SCOPE_PROCESS
  pthread_attr_setscope (&attr, PTHREAD_SCOPE_PROCESS);
#endif

  sigfillset (&fullsigset);

  pthread_sigmask (SIG_SETMASK, &fullsigset, &oldsigset);

  if (pthread_create (&pthrid, &attr, ping_proc, 0))
    croak ("AnyEvent::FastPing: unable to create pinger thread");

  pthread_sigmask (SIG_SETMASK, &oldsigset, 0);

  sv_setiv (get_sv ("AnyEvent::FastPing::THR_REQ_FD", 1), thr_send [1]);
  sv_setiv (get_sv ("AnyEvent::FastPing::THR_RES_FD", 1), thr_recv [0]);

  sv_setiv (get_sv ("AnyEvent::FastPing::ICMP4_FD", 1), icmp4_fd);
  sv_setiv (get_sv ("AnyEvent::FastPing::ICMP6_FD", 1), icmp6_fd);
}

MODULE = AnyEvent::FastPing             PACKAGE = AnyEvent::FastPing

BOOT:
{
  HV *stash = gv_stashpv ("AnyEvent::FastPing", 1);

  cbs = get_av ("AnyEvent::FastPing::CB", 1);
  magic = getpid () ^ MAGIC;

  boot ();

  newCONSTSUB (stash, "ipv4_supported", newSViv (icmp4_fd >= 0));
  newCONSTSUB (stash, "ipv6_supported", newSViv (icmp6_fd >= 0));

  newCONSTSUB (stash, "icmp4_pktsize", newSViv (HDR_SIZE_IP4 + sizeof (PKT)));
  newCONSTSUB (stash, "icmp6_pktsize", newSViv (HDR_SIZE_IP6 + sizeof (PKT)));
}

PROTOTYPES: DISABLE

SV *
_req_icmp_ping (SV *ranges, NV interval, U32 payload, SV *id)
        CODE:
{
        AV *rav;
        int nranges, i;
        REQ *req;

        if (!SvROK (ranges) || SvTYPE (SvRV (ranges)) != SVt_PVAV)
          croak ("address ranges must be given as arrayref with lo, hi pairs");

        rav = (AV *)SvRV (ranges);
        nranges = av_len (rav) + 1;

        req = malloc (sizeof (REQ));

        if (interval < MIN_INTERVAL)
          interval = MIN_INTERVAL;

        req->id       = newSVsv (id);
        req->interval = interval;
        req->payload  = payload;
        req->nranges  = nranges;
        req->ranges   = (RANGE *)malloc (nranges * sizeof (RANGE));

        while (nranges--)
          {
            SV *sv = *av_fetch (rav, nranges, 1);
            SV *lo, *hi;
            AV *av;
            RANGE *r;

            if (!SvROK (sv) || SvTYPE (SvRV (sv)) != SVt_PVAV)
              croak ("address range must be given as arrayref with lo, hi, interval arrayrefs");

            av = (AV *)SvRV (sv);
            r = req->ranges + nranges;

            lo = *av_fetch (av, 0, 1);
            hi = *av_fetch (av, 1, 1);

            sv_utf8_downgrade (lo, 0);
            sv_utf8_downgrade (hi, 0);

            memset (&r->lo, 0, sizeof (addr_t));
            memset (&r->hi, 0, sizeof (addr_t));

            if (SvPOKp (lo) && SvPOKp (hi))
              {
                if (SvCUR (lo) != SvCUR (hi))
                  croak ("all addresses in range must be of the same size (either 4 or 16 bytes)");

                if (SvCUR (lo) == 4)
                  {
                    r->family = AF_INET;
                    memcpy (sizeof (addr_t) - 4 + (char *)&r->lo, SvPVX (lo), 4);
                    memcpy (sizeof (addr_t) - 4 + (char *)&r->hi, SvPVX (hi), 4);
                  }
                else if (SvCUR (lo) == 16)
                  {
#if ENABLE_IPV6
                    r->family = AF_INET6;
                    memcpy (&r->lo, SvPVX (lo), sizeof (addr_t));
                    memcpy (&r->hi, SvPVX (hi), sizeof (addr_t));
#else
                    croak ("IPv6 not supported in this configuration");
#endif
                  }
                else
                  croak ("addresses in range must be either 4 (IPv4) or 16 (IPv6) bytes in length");
              }
            else if (SvIOK (lo) && SvIOK (hi))
              {
                uint32_t addr;

                r->family = AF_INET;

                addr = htonl (SvUV (lo)); memcpy (sizeof (addr_t) - 4 + (char *)&r->lo, &addr, 4);
                addr = htonl (SvUV (hi)); memcpy (sizeof (addr_t) - 4 + (char *)&r->hi, &addr, 4);
              }
            else
              croak ("addresses in range must be strings with either 4 (IPv4) or 16 (IPv6) octets");

            if (r->family == AF_INET)
              {
                if (icmp4_fd < 0)
                  croak ("AnyEvent::FastPing: IPv4 ping support not available on this system");
              }
            else
              {
                if (icmp6_fd < 0)
                  croak ("AnyEvent::FastPing: IPv6 ping support not available on this system");
              }

            r->interval = SvNV (*av_fetch (av, 2, 1));

            if (r->interval < req->interval)
              r->interval = req->interval;

            r->next = 0.;
          }

        RETVAL = newSVpvn ((char *)&req, sizeof (req));
}
        OUTPUT:
        RETVAL

SV *
_read_res ()
        CODE:
{
        REQ *req;

        if (read (thr_recv [0], &req, sizeof (req)) != sizeof (req))
          RETVAL = &PL_sv_undef;

        RETVAL = req->id;
        free (req->ranges);
        free (req);
}
        OUTPUT:
        RETVAL

void
_recv_icmp4 (...)
        CODE:
{
        char buf [512];
        struct sockaddr_in sa;
        socklen_t sl = sizeof (sa);
        AV *res_av = av_len (cbs) < 0 ? 0 : (AV *)sv_2mortal ((SV *)newAV ());
        tstamp now = NOW ();

        for (;;)
          {
            IP4HDR *iphdr = (IP4HDR *)buf;
            int len = recvfrom (icmp4_fd, buf, sizeof (buf), MSG_TRUNC, (struct sockaddr *)&sa, &sl);
            int hdrlen, totlen;
            PKT *pkt;

            if (len <= HDR_SIZE_IP4)
              break;

            hdrlen = (iphdr->version_ihl & 15) * 4;
            totlen = ntohs (iphdr->tot_len);

            // packet corrupt?
            if (!res_av
                || totlen > len
                || iphdr->protocol != IPPROTO_ICMP
                || hdrlen < HDR_SIZE_IP4 || hdrlen + sizeof (PKT) != totlen)
              continue;

            pkt = (PKT *)(buf + hdrlen);

            if (pkt->type != ICMP4_ECHO_REPLY
                || pkt->id  != (uint16_t) magic
                || pkt->seq != (uint16_t)~magic
                || !isnormal (pkt->stamp))
              continue;

            {
              AV *av = newAV ();
              av_push (av, newSVpvn ((char *)&sa.sin_addr, 4));
              av_push (av, newSVnv (now - pkt->stamp));
              av_push (av, newSVuv (pkt->payload));

              av_push (res_av, newRV_noinc ((SV *)av));
            }
          }

        if (res_av)
          feed_reply (res_av);
}

void
_recv_icmp6 (...)
        CODE:
{
        struct sockaddr_in6 sa;
        socklen_t sl = sizeof (sa);
        AV *res_av = av_len (cbs) < 0 ? 0 : (AV *)sv_2mortal ((SV *)newAV ());
        PKT pkt;
        tstamp now = NOW ();

        for (;;)
          {
            int len = recvfrom (icmp6_fd, &pkt, sizeof (pkt), MSG_TRUNC, (struct sockaddr *)&sa, &sl);

            if (len != sizeof (PKT))
              break;

            if (!res_av
                || pkt.type != ICMP6_ECHO_REPLY
                || pkt.id  != (uint16_t) magic
                || pkt.seq != (uint16_t)~magic
                || !isnormal (pkt.stamp))
              continue;

            {
              AV *av = newAV ();
              av_push (av, newSVpvn ((char *)&sa.sin6_addr, 16));
              av_push (av, newSVnv (now - pkt.stamp));
              av_push (av, newSVuv (pkt.payload));

              av_push (res_av, newRV_noinc ((SV *)av));
            }
          }

        if (res_av)
          feed_reply (res_av);
}

Revision:	1.6
Committed:	Sat Jul 18 00:53:42 2009 UTC (14 years, 10 months ago) by root
Branch:	MAIN
CVS Tags:	rel-1_13, rel-1_12
Changes since 1.5:	+51 -37 lines
Log Message:	* empty log message *
#	User	Rev	Content
1	root	1.5	#if defined(__linux)
2			# define ENABLE_IPV6 1 // if you get compilation problems try to disable IPv6
3	root	1.2	#else
4	root	1.5	# define ENABLE_IPV6 0
5	root	1.2	#endif
6	root	1.1
7			#include "EXTERN.h"
8			#include "perl.h"
9			#include "XSUB.h"
10
11			#include <pthread.h>
12
13			#include <math.h>
14			#include <stdio.h>
15			#include <stdlib.h>
16			#include <string.h>
17
18			#include <time.h>
19			#include <poll.h>
20			#include <unistd.h>
21			#include <inttypes.h>
22			#include <fcntl.h>
23			#include <errno.h>
24
25			#include <sys/types.h>
26			#include <sys/time.h>
27			#include <sys/socket.h>
28
29			#include <netinet/in.h>
30			#include <arpa/inet.h>
31
32			#ifdef __linux
33			# include <linux/icmp.h>
34			#endif
35	root	1.5	#if ENABLE_IPV6
36	root	1.1	# include <netinet/icmp6.h>
37			#endif
38
39			#define ICMP4_ECHO 8
40			#define ICMP4_ECHO_REPLY 0
41			#define ICMP6_ECHO 128
42			#define ICMP6_ECHO_REPLY 129
43
44			#define DRAIN_INTERVAL .000001 // how long to wait when sendto returns ENOBUFS, in seconds
45			#define MIN_INTERVAL .000001 // minimum packet send interval, in seconds
46
47			#define HDR_SIZE_IP4 20
48			#define HDR_SIZE_IP6 48
49
50	root	1.4	//TODO: xread/xwrite for atomicity? we currently rely on the fact that the pip buffersize divides exactly by pointer sizes
51	root	1.1
52			typedef uint8_t addr_t[16];
53
54			typedef double tstamp;
55
56	root	1.5	static tstamp
57			NOW (void)
58	root	1.1	{
59			struct timeval tv;
60			gettimeofday (&tv, 0);
61			return tv.tv_sec + tv.tv_usec * 0.000001;
62			}
63
64			typedef struct {
65			int family;
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;
81			uint8_t tos;
82			uint16_t tot_len;
83
84			uint16_t id;
85			uint16_t flags;
86
87			uint8_t ttl;
88			uint8_t protocol;
89			uint16_t cksum;
90
91			uint32_t src;
92			uint32_t dst;
93			} IP4HDR;
94
95			typedef struct {
96			uint8_t version;
97			uint8_t x1, x2, x3;
98
99			uint16_t payload_len;
100			uint8_t nxt_hdr;
101			uint8_t hop_limit;
102
103			uint8_t src[16];
104			uint8_t dst[16];
105			} IP6HDR;
106
107			#define MAGIC 0xca4c
108
109			static uint16_t magic;
110
111			typedef struct {
112			uint8_t type, code;
113			uint16_t cksum;
114			uint16_t id, seq;
115			uint32_t payload;
116			tstamp stamp; // be careful when accessing this
117			} PKT;
118
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
128			icmp_cksum (void *data, unsigned int len)
129			{
130			register int sum = 0;
131			uint16_t *wp;
132
133			assert (~len & 1);
134
135			for (wp = (uint16_t *)data; len; wp++, len -= 2)
136			sum += *wp;
137
138			sum = (sum >> 16) + (sum & 0xffff); /* add high 16 to low 16 */
139			sum += sum >> 16; /* add carry */
140
141			return ~sum;
142			}
143
144			static void
145			inc_addr (addr_t *addr)
146			{
147			int len = sizeof (addr_t) - 1;
148
149			do
150			{
151			if ((*addr)[len] != 0xff)
152			{
153			++(*addr)[len];
154			break;
155			}
156
157			(*addr)[len] = 0;
158			}
159			while (len--);
160			}
161
162			static void *
163			ping_proc (void *unused)
164			{
165			PKT pkt;
166			struct sockaddr_in sa4;
167	root	1.5	#if ENABLE_IPV6
168	root	1.1	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	root	1.5	#if ENABLE_IPV6
177	root	1.1	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	root	1.6	tstamp now = NOW ();
188			tstamp next = now;
189
190	root	1.1	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			}
210
211			while (req->nranges)
212			{
213			RANGE *range = req->ranges;
214	root	1.6	int n, k;
215	root	1.1
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	root	1.5	#if ENABLE_IPV6
263	root	1.1	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	root	1.6	for (n = k = 0; ; )
296	root	1.1	{
297	root	1.6	int j = k * 2 + 1;
298
299	root	1.1	++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	root	1.6	{
311			RANGE temp = req->ranges [k];
312			req->ranges [k] = req->ranges [j];
313			req->ranges [j] = temp;
314			}
315	root	1.1
316			k = j;
317			}
318			}
319
320			write (thr_recv [1], &req, sizeof (req));
321			}
322
323			return 0;
324			}
325
326			static void
327			feed_reply (AV *res_av)
328			{
329			dSP;
330			SV res = sv_2mortal (newRV_inc ((SV )res_av));
331			int i;
332
333	root	1.6	if (av_len (res_av) < 0)
334			return;
335
336	root	1.1	ENTER;
337			SAVETMPS;
338
339			for (i = av_len (cbs) + 1; i--; )
340			{
341			SV cb = av_fetch (cbs, i, 1);
342
343			PUSHMARK (SP);
344			XPUSHs (res);
345			PUTBACK;
346			call_sv (cb, G_DISCARD \| G_VOID);
347			}
348
349			FREETMPS;
350			LEAVE;
351			}
352
353			static void
354			boot ()
355			{
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)
363			croak ("AnyEvent::FastPing: unable to create receive pipe");
364
365			icmp4_fd = socket (AF_INET, SOCK_RAW, IPPROTO_ICMP);
366	root	1.2	fcntl (icmp4_fd, F_SETFL, O_NONBLOCK);
367	root	1.1	#ifdef ICMP_FILTER
368			{
369			struct icmp_filter oval;
370			oval.data = 0xffffffff & ~(1 << ICMP4_ECHO_REPLY);
371			setsockopt (icmp4_fd, SOL_RAW, ICMP_FILTER, &oval, sizeof oval);
372			}
373			#endif
374
375	root	1.5	#if ENABLE_IPV6
376	root	1.1	icmp6_fd = socket (AF_INET6, SOCK_RAW, IPPROTO_ICMPV6);
377	root	1.2	fcntl (icmp6_fd, F_SETFL, O_NONBLOCK);
378	root	1.1	# ifdef ICMP6_FILTER
379			{
380			struct icmp6_filter oval;
381			ICMP6_FILTER_SETBLOCKALL (&oval);
382			ICMP6_FILTER_SETPASS (ICMP6_ECHO_REPLY, &oval);
383			setsockopt (icmp6_fd, IPPROTO_ICMPV6, ICMP6_FILTER, &oval, sizeof oval);
384			}
385			# endif
386			#endif
387
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);
407			sv_setiv (get_sv ("AnyEvent::FastPing::ICMP6_FD", 1), icmp6_fd);
408			}
409
410			MODULE = AnyEvent::FastPing PACKAGE = AnyEvent::FastPing
411
412			BOOT:
413			{
414			HV *stash = gv_stashpv ("AnyEvent::FastPing", 1);
415
416			cbs = get_av ("AnyEvent::FastPing::CB", 1);
417			magic = getpid () ^ MAGIC;
418
419			boot ();
420
421			newCONSTSUB (stash, "ipv4_supported", newSViv (icmp4_fd >= 0));
422			newCONSTSUB (stash, "ipv6_supported", newSViv (icmp6_fd >= 0));
423
424			newCONSTSUB (stash, "icmp4_pktsize", newSViv (HDR_SIZE_IP4 + sizeof (PKT)));
425			newCONSTSUB (stash, "icmp6_pktsize", newSViv (HDR_SIZE_IP6 + sizeof (PKT)));
426			}
427
428			PROTOTYPES: DISABLE
429
430			SV *
431			_req_icmp_ping (SV ranges, NV interval, U32 payload, SV id)
432			CODE:
433			{
434	root	1.6	AV *rav;
435			int nranges, i;
436			REQ *req;
437
438	root	1.1	if (!SvROK (ranges) \|\| SvTYPE (SvRV (ranges)) != SVt_PVAV)
439			croak ("address ranges must be given as arrayref with lo, hi pairs");
440
441	root	1.6	rav = (AV *)SvRV (ranges);
442			nranges = av_len (rav) + 1;
443	root	1.1
444	root	1.6	req = malloc (sizeof (REQ));
445	root	1.1
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	root	1.6	SV lo, hi;
459			AV *av;
460			RANGE *r;
461	root	1.1
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	root	1.6	av = (AV *)SvRV (sv);
466			r = req->ranges + nranges;
467	root	1.1
468	root	1.6	lo = *av_fetch (av, 0, 1);
469			hi = *av_fetch (av, 1, 1);
470	root	1.1
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	root	1.5	croak ("all addresses in range must be of the same size (either 4 or 16 bytes)");
481	root	1.1
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	root	1.5	#if ENABLE_IPV6
491	root	1.1	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	root	1.5	croak ("addresses in range must be either 4 (IPv4) or 16 (IPv6) bytes in length");
500	root	1.1	}
501			else if (SvIOK (lo) && SvIOK (hi))
502			{
503	root	1.6	uint32_t addr;
504
505	root	1.1	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
554			_recv_icmp4 (...)
555			CODE:
556			{
557			char buf [512];
558			struct sockaddr_in sa;
559			socklen_t sl = sizeof (sa);
560			AV res_av = av_len (cbs) < 0 ? 0 : (AV )sv_2mortal ((SV *)newAV ());
561			tstamp now = NOW ();
562
563			for (;;)
564			{
565	root	1.6	IP4HDR iphdr = (IP4HDR )buf;
566	root	1.2	int len = recvfrom (icmp4_fd, buf, sizeof (buf), MSG_TRUNC, (struct sockaddr *)&sa, &sl);
567	root	1.6	int hdrlen, totlen;
568			PKT *pkt;
569	root	1.1
570			if (len <= HDR_SIZE_IP4)
571			break;
572
573	root	1.6	hdrlen = (iphdr->version_ihl & 15) * 4;
574			totlen = ntohs (iphdr->tot_len);
575	root	1.1
576			// packet corrupt?
577			if (!res_av
578			\|\| totlen > len
579			\|\| iphdr->protocol != IPPROTO_ICMP
580			\|\| hdrlen < HDR_SIZE_IP4 \|\| hdrlen + sizeof (PKT) != totlen)
581			continue;
582
583	root	1.6	pkt = (PKT *)(buf + hdrlen);
584	root	1.1
585			if (pkt->type != ICMP4_ECHO_REPLY
586			\|\| pkt->id != (uint16_t) magic
587			\|\| pkt->seq != (uint16_t)~magic
588			\|\| !isnormal (pkt->stamp))
589			continue;
590
591	root	1.6	{
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	root	1.1
597	root	1.6	av_push (res_av, newRV_noinc ((SV *)av));
598			}
599	root	1.1	}
600
601			if (res_av)
602			feed_reply (res_av);
603			}
604
605			void
606			_recv_icmp6 (...)
607			CODE:
608			{
609			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;
613			tstamp now = NOW ();
614
615			for (;;)
616			{
617	root	1.2	int len = recvfrom (icmp6_fd, &pkt, sizeof (pkt), MSG_TRUNC, (struct sockaddr *)&sa, &sl);
618	root	1.1
619			if (len != sizeof (PKT))
620			break;
621
622			if (!res_av
623			\|\| pkt.type != ICMP6_ECHO_REPLY
624			\|\| pkt.id != (uint16_t) magic
625			\|\| pkt.seq != (uint16_t)~magic
626			\|\| !isnormal (pkt.stamp))
627			continue;
628
629	root	1.6	{
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	root	1.1
635	root	1.6	av_push (res_av, newRV_noinc ((SV *)av));
636			}
637	root	1.1	}
638
639			if (res_av)
640			feed_reply (res_av);
641			}
642