AnyEvent-FastPing/FastPing.xs

#if defined(__linux) || defined(__FreeBSD__) || defined(__OpenBSD__) || defined(__NetBSD__) || defined(__CYGWIN__)
# 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 <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 && !defined (__CYGWIN__)
# 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 1e-6 // how long to wait when sendto returns ENOBUFS, in seconds
#define MIN_INTERVAL   1e-6 // minimum packet send interval, in seconds

#define HDR_SIZE_IP4  20
#define HDR_SIZE_IP6  48

static int thr_res[2]; // worker thread finished status
static int icmp4_fd = -1;
static int icmp6_fd = -1;

/*****************************************************************************/

typedef double tstamp;

static tstamp
NOW (void)
{
  struct timeval tv;

  gettimeofday (&tv, 0);

  return tv.tv_sec + tv.tv_usec * 1e-6;
}

static void
ssleep (tstamp wait)
{
#if defined (__SVR4) && defined (__sun)
  struct timeval tv;

  tv.tv_sec  = wait;
  tv.tv_usec = (wait - tv.tv_sec) * 1e6;

  select (0, 0, 0, 0, &tv);
#elif defined(_WIN32)
  Sleep ((unsigned long)(delay * 1e3));
#else
  struct timespec ts;

  ts.tv_sec  = wait;
  ts.tv_nsec = (wait - ts.tv_sec) * 1e9;

  nanosleep (&ts, 0);
#endif
}

/*****************************************************************************/

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 uint8_t addr_tt[16];

typedef struct
{
  tstamp next;
  double interval;
  int addrlen;

  addr_tt lo, hi; /* only if !addrcnt */

  int addrcnt;
  /* addrcnt addresses follow */
} RANGE;

typedef struct
{
  RANGE **ranges;
  int rangecnt, rangemax;

  tstamp interval;
  tstamp maxrtt;
  uint16_t magic1;
  uint16_t magic2;
  uint16_t magic3;

  int id;

  AV *recvq; /* receive queue */
  int nextrecv;
  SV *recvcb;

  pthread_t thrid;
  int running;
} PINGER;

static PINGER **pingers;
static int *pingerfree; /* freelist next */
static int pingercnt;
static int pingermax;
static int firstfree = -1;
static int firstrecv = -1;

/*****************************************************************************/

typedef struct
{
  uint8_t type, code;
  uint16_t cksum;

  uint16_t id, seq;

  uint16_t pinger;
  uint16_t magic;

  uint32_t stamp_hi;
  uint32_t stamp_lo;
} PKT;

static int
pkt_is_valid_for (PKT *pkt, PINGER *pinger)
{
  return pkt->id    == pinger->magic1
      && pkt->seq   == pinger->magic2
      && pkt->magic == pinger->magic3;
}

static void
ts_to_pkt (PKT *pkt, tstamp ts)
{
  /* move 12 bits of seconds into the 32 bit fractional part */
  /* leaving 20 bits subsecond resolution and 44 bits of integers */
  /* (of which 32 are typically usable) */
  ts *= 1. / 4096.;

  pkt->stamp_hi = ts;
  pkt->stamp_lo = (ts - pkt->stamp_hi) * 4294967296.;
}

static tstamp
pkt_to_ts (PKT *pkt)
{
  return pkt->stamp_hi *  4096.
       + pkt->stamp_lo * (4096. / 4294967296.);
}

static void
pkt_cksum (PKT *pkt)
{
  uint_fast32_t sum = -pkt->cksum;
  uint32_t *wp = (uint32_t *)pkt;
  int len = sizeof (*pkt) / 4;

  do
    {
      uint_fast32_t w = *(volatile uint32_t *)wp++;
      sum += (w & 0xffff) + (w >> 16);
    }
  while (len--);

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

  pkt->cksum = ~sum;
}

/*****************************************************************************/

static void
range_free (RANGE *self)
{
  free (self);
}

/* like sendto, but retries on failure */
static void
xsendto (int fd, void *buf, size_t len, int flags, void *sa, int salen)
{
  tstamp wait = DRAIN_INTERVAL / 2.;

  while (sendto (fd, buf, len, flags, sa, salen) < 0 && errno == ENOBUFS)
    ssleep (wait *= 2.);
}

// ping current address, return true and increment if more to ping
static int
range_send_ping (RANGE *self, PKT *pkt)
{
  // send ping
  uint8_t *addr;

  if (self->addrcnt)
    addr = (self->addrcnt - 1) * self->addrlen + (uint8_t *)(self + 1);
  else
    addr = sizeof (addr_tt) - self->addrlen + self->lo;

  pkt->cksum = 0;

  if (self->addrlen == 4)
    {
      struct sockaddr_in sa;

      pkt->type  = ICMP4_ECHO;
      pkt_cksum (pkt);

      sa.sin_family = AF_INET;
      sa.sin_port   = 0;

      memcpy (&sa.sin_addr, addr, sizeof (sa.sin_addr));

      xsendto (icmp4_fd, pkt, sizeof (*pkt), 0, &sa, sizeof (sa));
    }
  else
    {
#if ENABLE_IPV6
      struct sockaddr_in6 sa;

      pkt->type = ICMP6_ECHO;

      sa.sin6_family   = AF_INET6;
      sa.sin6_port     = 0;
      sa.sin6_flowinfo = 0;
      sa.sin6_scope_id = 0;

      memcpy (&sa.sin6_addr, addr, sizeof (sa.sin6_addr));

      xsendto (icmp6_fd, pkt, sizeof (*pkt), 0, &sa, sizeof (sa));
#endif
    }

  // see if we have any more addresses
  if (self->addrcnt)
    {
      if (!--self->addrcnt)
        return 0;
    }
  else
    {
      if (!memcmp (&self->lo, &self->hi, sizeof (addr_tt)))
        return 0;

      // increment self->lo
      {
        int len = sizeof (addr_tt) - 1;

        while (!++self->lo [len])
          --len;
      }
    }

  return 1;
}

/*****************************************************************************/

static void
downheap (PINGER *self)
{
  RANGE *elem = self->ranges [0]; /* always exists */
  int Nm1 = self->rangecnt - 1;
  int j;
  int k;

  for (k = 0; ; )
    {
      int j = k * 2 + 1;

      if (j > Nm1)
        break;

      if (j < Nm1
          && self->ranges [j]->next > self->ranges [j + 1]->next)
        ++j;

      if (self->ranges [j]->next >= elem->next)
        break;

      self->ranges [k] = self->ranges [j];

      k = j;
    }

  self->ranges [k] = elem;
}

static void
upheap (PINGER *self, int k)
{
  RANGE *elem = self->ranges [k];

  while (k)
    {
      int j = (k - 1) >> 1;

      if (self->ranges [j]->next <= elem->next)
        break;

      self->ranges [k] = self->ranges [j];

      k = j;
    }

  self->ranges [k] = elem;
}

static void *
ping_proc (void *self_)
{
  PINGER *self = (PINGER *)self_;
  PKT pkt;

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

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

  pkt.code    = 0;
  pkt.id      = self->magic1;
  pkt.seq     = self->magic2;
  pkt.magic   = self->magic3;
  pkt.pinger  = self->id;

  while (self->rangecnt)
    {
      RANGE *range = self->ranges [0];
      int n, k;

      // 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.)
        ssleep (wait);

      now = NOW ();

      ts_to_pkt (&pkt, now);

      if (!range_send_ping (range, &pkt))
        {
          self->ranges [0] = self->ranges [--self->rangecnt];
          range_free (range);
        }
      else
        {
          range->next = next;
          range->next += range->interval;
        }

      downheap (self);

      next += self->interval;
    }

  ssleep (self->maxrtt);

  {
    uint16_t id = self->id;

    write (thr_res [1], &id, sizeof (id));
  }

  return 0;
}

/*****************************************************************************/

static void
pinger_start (PINGER *self)
{
  sigset_t fullsigset, oldsigset;
  pthread_attr_t attr;

  if (self->running)
    return;

  sigfillset (&fullsigset);

  pthread_attr_init (&attr);
  pthread_attr_setstacksize (&attr, PTHREAD_STACK_MIN < sizeof (long) * 2048 ? sizeof (long) * 2048 : PTHREAD_STACK_MIN);

  pthread_sigmask (SIG_SETMASK, &fullsigset, &oldsigset);

  if (pthread_create (&self->thrid, &attr, ping_proc, (void *)self))
    croak ("AnyEvent::FastPing: unable to create pinger thread");

  pthread_sigmask (SIG_SETMASK, &oldsigset, 0);

  self->running = 1;
}

static void
pinger_stop (PINGER *self)
{
  if (!self->running)
    return;

  self->running = 0;
  pthread_cancel (self->thrid);
  pthread_join (self->thrid, 0);
}

static void
pinger_init (PINGER *self)
{
  memset (self, 0, sizeof (PINGER));

  if (firstfree >= 0)
    {
      self->id = firstfree;
      firstfree = pingerfree [firstfree];
    }
  else if (pingercnt == 0xffff)
    croak ("unable to create more than 65536 AnyEvent::FastPing objects");
  else
    {
      if (pingercnt == pingermax)
        {
          pingermax = pingermax * 2 + 16;
          pingers    = realloc (pingers   , sizeof (pingers    [0]) * pingermax);
          pingerfree = realloc (pingerfree, sizeof (pingerfree [0]) * pingermax);
        }

      self->id = pingercnt++;
    }

  pingers [self->id] = self;

  self->recvcb   = &PL_sv_undef;
  self->interval = MIN_INTERVAL;
  self->maxrtt   = 0.5;
  self->rangemax = 16;
  self->ranges   = malloc (sizeof (self->ranges [0]) * self->rangemax);
}

static void
pinger_free (PINGER *self)
{
  pinger_stop (self);

  pingers [self->id] = 0;

  SvREFCNT_dec (self->recvq);
  SvREFCNT_dec (self->recvcb);

  pingerfree [self->id] = firstfree;
  firstfree = self->id;

  while (self->rangecnt)
    range_free (self->ranges [--self->rangecnt]);

  free (self->ranges);
}

static void
pinger_add_range (PINGER *self, RANGE *range)
{
  if (self->rangecnt == self->rangemax)
    self->ranges = realloc (self->ranges, sizeof (self->ranges [0]) * (self->rangemax <<= 1));

  self->ranges [self->rangecnt] = range;
  upheap (self, self->rangecnt);
  ++self->rangecnt;
}

/*****************************************************************************/

static void
recv_feed (PINGER *self, void *addr, int addrlen, tstamp rtt)
{
  if (!self->recvq)
    {
      /* first seen this round */
      if (!SvOK (self->recvcb))
        return;

      self->recvq = newAV ();

      self->nextrecv = firstrecv;
      firstrecv = self->id;
    }

  {
    AV *pkt = newAV ();

    av_extend (pkt, 2-1);

    AvARRAY (pkt)[0] = newSVpvn (addr, addrlen);
    AvARRAY (pkt)[1] = newSVnv (rtt);
    AvFILLp (pkt) = 2-1;

    av_push (self->recvq, newRV_noinc ((SV *)pkt));
  }
}

static void
recv_flush (void)
{
  if (firstrecv < 0)
    return;

  ENTER;
  SAVETMPS;

  do
    {
      dSP;
      PINGER *self = pingers [firstrecv];
      firstrecv = self->nextrecv;

      self->nextrecv = -1;

      PUSHMARK (SP);
      XPUSHs (sv_2mortal (newRV_noinc ((SV *)self->recvq)));
      self->recvq = 0;
      PUTBACK;
      call_sv (self->recvcb, G_DISCARD | G_VOID);
    }
  while (firstrecv >= 0);

  FREETMPS;
  LEAVE;
}

/*****************************************************************************/

#if 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;
}
#endif

static void
boot ()
{
  if (pipe (thr_res) < 0)
    croak ("AnyEvent::FastPing: unable to create receive pipe");

  sv_setiv (get_sv ("AnyEvent::FastPing::THR_RES_FD", 1), thr_res [0]);

  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

  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            PREFIX = pinger_

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

  if (sizeof (PKT) & 3)
    croak ("size of PKT structure is not a multiple of 4");

  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

void
_recv_icmp4 (...)
        CODE:
{
        char buf [512];
        struct sockaddr_in sa;
        int maxrecv;

        for (maxrecv = 256+1; --maxrecv; )
          {
            PINGER *pinger;
            IP4HDR *iphdr = (IP4HDR *)buf;
            socklen_t sl = sizeof (sa);
            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);

            if (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->pinger >= pingercnt
                || !pingers [pkt->pinger])
              continue;

            pinger = pingers [pkt->pinger];

            if (!pkt_is_valid_for (pkt, pinger))
              continue;

            recv_feed (pinger, &sa.sin_addr, 4, NOW () - pkt_to_ts (pkt));
          }

        recv_flush ();
}

void
_recv_icmp6 (...)
        CODE:
{
        struct sockaddr_in6 sa;
        PKT pkt;
        int maxrecv;

        for (maxrecv = 256+1; --maxrecv; )
          {
            PINGER *pinger;
            socklen_t sl = sizeof (sa);
            int len = recvfrom (icmp6_fd, &pkt, sizeof (pkt), MSG_TRUNC, (struct sockaddr *)&sa, &sl);

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

            if (pkt.type != ICMP6_ECHO_REPLY
                || pkt.pinger >= pingercnt
                || !pingers [pkt.pinger])
              continue;

            pinger = pingers [pkt.pinger];

            if (!pkt_is_valid_for (&pkt, pinger))
              continue;

            recv_feed (pinger, &sa.sin6_addr, 16, NOW () - pkt_to_ts (&pkt));
          }

        recv_flush ();
}

void
_new (SV *klass, UV magic1, UV magic2, UV magic3)
        PPCODE:
{
        SV *pv = NEWSV (0, sizeof (PINGER));
        PINGER *self = (PINGER *)SvPVX (pv);
        SvPOK_only (pv);
        XPUSHs (sv_2mortal (sv_bless (newRV_noinc (pv), gv_stashpv (SvPVutf8_nolen (klass), 1))));
        pinger_init (self);
        self->magic1 = magic1;
        self->magic2 = magic2;
        self->magic3 = magic3;
}

void
_free (PINGER *self)
        CODE:
        pinger_free (self);

IV
id (PINGER *self, ...)
        CODE:
        RETVAL = self->id;
        OUTPUT:
        RETVAL

void pinger_start (PINGER *self)

void pinger_stop (PINGER *self)

void
_stop_id (UV id)
        CODE:
        if (id < pingercnt && pingers [id])
          pinger_stop (pingers [id]);

void
interval (PINGER *self, NV interval)
        CODE:
        self->interval = interval > MIN_INTERVAL ? interval : MIN_INTERVAL;

void
max_rtt (PINGER *self, NV maxrtt)
        CODE:
        self->maxrtt = maxrtt;

void
add_range (PINGER *self, SV *lo_, SV *hi_, NV interval)
        CODE:
{
        STRLEN lo_len, hi_len;
        char *lo = SvPVbyte (lo_, lo_len);
        char *hi = SvPVbyte (hi_, hi_len);
        RANGE *range;

        if (lo_len != hi_len || (lo_len != 4 && lo_len != 16))
          croak ("AnyEvent::FastPing::add_range address range must be specified as two binary IPv4 or IPv6 addresses");

        if (lo_len ==  4 && icmp4_fd < 0) croak ("IPv4 support unavailable");
        if (lo_len == 16 && icmp6_fd < 0) croak ("IPv6 support unavailable");

        if (memcmp (lo, hi, lo_len) > 0)
          croak ("AnyEvent::FastPing::add_range called with lo > hi");

        range = calloc (1, sizeof (RANGE));

        range->next     = 0;
        range->interval = interval > MIN_INTERVAL ? interval : MIN_INTERVAL;
        range->addrlen  = lo_len;

        memcpy (sizeof (addr_tt) - lo_len + (char *)&range->lo, lo, lo_len);
        memcpy (sizeof (addr_tt) - lo_len + (char *)&range->hi, hi, lo_len);

        pinger_add_range (self, range);
}

void
add_hosts (PINGER *self, SV *addrs, NV interval)
        CODE:
{
        AV *av;
        int i;
        int cnt;
        int addrlen;
        RANGE *range;

        if (!SvROK (addrs) || SvTYPE (SvRV (addrs)) != SVt_PVAV)
          croak ("AnyEvent::FastPing::add_hosst expects an arrayref with binary IPv4 or IPv6 addresses");

        av = (AV *)SvRV (addrs);
        cnt = av_len (av) + 1;

        if (!cnt)
          XSRETURN_EMPTY;

        addrlen = SvCUR (*av_fetch (av, 0, 1));

        if (addrlen != 4 && addrlen != 16)
          croak ("AnyEvent::FastPing::add_hosts addresses must be specified as binary IPv4 or IPv6 addresses");

        for (i = cnt; --i; )
          {
            SV *sv = *av_fetch (av, i, 1);

            if (!sv_utf8_downgrade (sv, 1) || addrlen != SvCUR (sv))
              croak ("AnyEvent::FastPing::add_hosts addresses must all have the same size");
          }

        range = calloc (1, sizeof (RANGE) + cnt * addrlen);

        range->next     = 0;
        range->interval = interval > MIN_INTERVAL ? interval : MIN_INTERVAL;
        range->addrlen  = addrlen;
        range->addrcnt  = cnt;

        for (i = cnt; i--; )
          memcpy ((uint8_t *)(range + 1) + (cnt - 1 - i) * addrlen,
                  SvPVbyte_nolen (*av_fetch (av, i, 1)),
                  addrlen);

        pinger_add_range (self, range);
}

void
on_recv (PINGER *self, SV *cb)
        CODE:
        SvREFCNT_dec (self->recvcb);
        self->recvcb = newSVsv (cb);

Revision:	1.11
Committed:	Tue Feb 1 04:06:24 2011 UTC (13 years, 3 months ago) by root
Branch:	MAIN
Changes since 1.10:	+200 -135 lines
Log Message:	* empty log message *
#	User	Rev	Content
1	root	1.8	#if defined(__linux) \|\| defined(__FreeBSD__) \|\| defined(__OpenBSD__) \|\| defined(__NetBSD__) \|\| defined(__CYGWIN__)
2	root	1.5	# 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 <stdio.h>
14			#include <stdlib.h>
15			#include <string.h>
16
17			#include <time.h>
18			#include <poll.h>
19			#include <unistd.h>
20			#include <inttypes.h>
21			#include <fcntl.h>
22			#include <errno.h>
23
24			#include <sys/types.h>
25			#include <sys/time.h>
26			#include <sys/socket.h>
27
28			#include <netinet/in.h>
29			#include <arpa/inet.h>
30
31			#ifdef __linux
32			# include <linux/icmp.h>
33			#endif
34	root	1.8	#if ENABLE_IPV6 && !defined (__CYGWIN__)
35	root	1.1	# include <netinet/icmp6.h>
36			#endif
37
38			#define ICMP4_ECHO 8
39			#define ICMP4_ECHO_REPLY 0
40			#define ICMP6_ECHO 128
41			#define ICMP6_ECHO_REPLY 129
42
43	root	1.9	#define DRAIN_INTERVAL 1e-6 // how long to wait when sendto returns ENOBUFS, in seconds
44			#define MIN_INTERVAL 1e-6 // minimum packet send interval, in seconds
45	root	1.1
46			#define HDR_SIZE_IP4 20
47			#define HDR_SIZE_IP6 48
48
49	root	1.11	static int thr_res[2]; // worker thread finished status
50			static int icmp4_fd = -1;
51			static int icmp6_fd = -1;
52	root	1.10
53	root	1.9	/*****************************************************************************/
54
55	root	1.1	typedef double tstamp;
56
57	root	1.5	static tstamp
58			NOW (void)
59	root	1.1	{
60			struct timeval tv;
61	root	1.9
62	root	1.1	gettimeofday (&tv, 0);
63	root	1.9
64			return tv.tv_sec + tv.tv_usec * 1e-6;
65	root	1.1	}
66
67	root	1.9	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
91	root	1.8	typedef struct
92			{
93	root	1.1	uint8_t version_ihl;
94			uint8_t tos;
95			uint16_t tot_len;
96
97			uint16_t id;
98			uint16_t flags;
99
100			uint8_t ttl;
101			uint8_t protocol;
102			uint16_t cksum;
103
104			uint32_t src;
105			uint32_t dst;
106			} IP4HDR;
107
108	root	1.11	/*****************************************************************************/
109	root	1.1
110	root	1.11	typedef uint8_t addr_tt[16];
111	root	1.9
112	root	1.10	typedef struct
113			{
114			tstamp next;
115			double interval;
116			int addrlen;
117	root	1.11
118			addr_tt lo, hi; /* only if !addrcnt */
119
120			int addrcnt;
121			/* addrcnt addresses follow */
122	root	1.10	} RANGE;
123
124			typedef struct
125			{
126			RANGE **ranges;
127			int rangecnt, rangemax;
128	root	1.1
129	root	1.10	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	root	1.1
154	root	1.8	typedef struct
155			{
156	root	1.1	uint8_t type, code;
157			uint16_t cksum;
158	root	1.11
159	root	1.1	uint16_t id, seq;
160	root	1.11
161	root	1.10	uint16_t pinger;
162			uint16_t magic;
163	root	1.11
164	root	1.9	uint32_t stamp_hi;
165			uint32_t stamp_lo;
166	root	1.1	} PKT;
167
168	root	1.9	static int
169	root	1.10	pkt_is_valid_for (PKT pkt, PINGER pinger)
170	root	1.9	{
171	root	1.10	return pkt->id == pinger->magic1
172			&& pkt->seq == pinger->magic2
173			&& pkt->magic == pinger->magic3;
174	root	1.9	}
175
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.;
183
184			pkt->stamp_hi = ts;
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	root	1.10	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--);
208
209			sum = (sum >> 16) + (sum & 0xffff); /* add high 16 to low 16 */
210			sum += sum >> 16; /* add carry */
211
212			pkt->cksum = ~sum;
213			}
214
215	root	1.9	/*****************************************************************************/
216
217	root	1.10	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	root	1.1	{
227	root	1.11	tstamp wait = DRAIN_INTERVAL / 2.;
228	root	1.10
229			while (sendto (fd, buf, len, flags, sa, salen) < 0 && errno == ENOBUFS)
230	root	1.11	ssleep (wait *= 2.);
231	root	1.10	}
232
233	root	1.11	// ping current address, return true and increment if more to ping
234			static int
235			range_send_ping (RANGE self, PKT pkt)
236	root	1.10	{
237	root	1.11	// 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	root	1.10	pkt->cksum = 0;
246
247	root	1.11	if (self->addrlen == 4)
248	root	1.10	{
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	root	1.11	memcpy (&sa.sin_addr, addr, sizeof (sa.sin_addr));
258	root	1.10
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	root	1.1
268	root	1.10	sa.sin6_family = AF_INET6;
269			sa.sin6_port = 0;
270			sa.sin6_flowinfo = 0;
271			sa.sin6_scope_id = 0;
272	root	1.1
273	root	1.11	memcpy (&sa.sin6_addr, addr, sizeof (sa.sin6_addr));
274	root	1.10
275	root	1.11	xsendto (icmp6_fd, pkt, sizeof (*pkt), 0, &sa, sizeof (sa));
276	root	1.10	#endif
277			}
278	root	1.11
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
299			return 1;
300	root	1.10	}
301
302	root	1.11	/*****************************************************************************/
303
304	root	1.10	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	root	1.1
331	root	1.10	self->ranges [k] = elem;
332	root	1.1	}
333
334			static void
335	root	1.10	upheap (PINGER *self, int k)
336	root	1.1	{
337	root	1.10	RANGE *elem = self->ranges [k];
338	root	1.1
339	root	1.10	while (k)
340	root	1.1	{
341	root	1.10	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	root	1.1
348	root	1.10	k = j;
349	root	1.1	}
350	root	1.10
351			self->ranges [k] = elem;
352	root	1.1	}
353
354			static void *
355	root	1.10	ping_proc (void *self_)
356	root	1.1	{
357	root	1.10	PINGER self = (PINGER )self_;
358	root	1.1	PKT pkt;
359
360			memset (&pkt, 0, sizeof (pkt));
361
362	root	1.10	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	root	1.1
371	root	1.10	while (self->rangecnt)
372	root	1.1	{
373	root	1.10	RANGE *range = self->ranges [0];
374			int n, k;
375	root	1.1
376	root	1.10	// ranges [0] is always the next range to ping
377			tstamp wait = range->next - now;
378	root	1.6
379	root	1.10	// 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	root	1.11	if (!range_send_ping (range, &pkt))
397	root	1.1	{
398	root	1.10	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	root	1.11	}
406	root	1.10
407	root	1.11	downheap (self);
408	root	1.1
409	root	1.10	next += self->interval;
410			}
411
412			ssleep (self->maxrtt);
413
414			{
415			uint16_t id = self->id;
416	root	1.1
417	root	1.10	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	root	1.1
436	root	1.10	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	root	1.1
465	root	1.10	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	root	1.1	{
476	root	1.10	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	root	1.11	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	root	1.10	/*****************************************************************************/
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	root	1.1
540	root	1.10	{
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	root	1.1
549	root	1.10	av_push (self->recvq, newRV_noinc ((SV *)pkt));
550			}
551			}
552	root	1.1
553	root	1.10	static void
554			recv_flush (void)
555			{
556			if (firstrecv < 0)
557			return;
558
559			ENTER;
560			SAVETMPS;
561	root	1.1
562	root	1.11	do
563	root	1.10	{
564			dSP;
565			PINGER *self = pingers [firstrecv];
566			firstrecv = self->nextrecv;
567	root	1.1
568	root	1.10	self->nextrecv = -1;
569	root	1.1
570	root	1.10	PUSHMARK (SP);
571			XPUSHs (sv_2mortal (newRV_noinc ((SV *)self->recvq)));
572			self->recvq = 0;
573			PUTBACK;
574			call_sv (self->recvcb, G_DISCARD \| G_VOID);
575	root	1.1	}
576	root	1.11	while (firstrecv >= 0);
577	root	1.1
578	root	1.10	FREETMPS;
579			LEAVE;
580	root	1.1	}
581
582	root	1.10	/*****************************************************************************/
583
584			#if 0
585	root	1.1	static void
586			feed_reply (AV *res_av)
587			{
588			dSP;
589			SV res = sv_2mortal (newRV_inc ((SV )res_av));
590			int i;
591
592	root	1.6	if (av_len (res_av) < 0)
593			return;
594
595	root	1.1	ENTER;
596			SAVETMPS;
597
598			for (i = av_len (cbs) + 1; i--; )
599			{
600			SV cb = av_fetch (cbs, i, 1);
601
602			PUSHMARK (SP);
603			XPUSHs (res);
604			PUTBACK;
605			call_sv (cb, G_DISCARD \| G_VOID);
606			}
607
608			FREETMPS;
609			LEAVE;
610			}
611	root	1.10	#endif
612	root	1.1
613			static void
614			boot ()
615			{
616	root	1.10	if (pipe (thr_res) < 0)
617			croak ("AnyEvent::FastPing: unable to create receive pipe");
618	root	1.1
619	root	1.10	sv_setiv (get_sv ("AnyEvent::FastPing::THR_RES_FD", 1), thr_res [0]);
620	root	1.1
621			icmp4_fd = socket (AF_INET, SOCK_RAW, IPPROTO_ICMP);
622	root	1.2	fcntl (icmp4_fd, F_SETFL, O_NONBLOCK);
623	root	1.1	#ifdef ICMP_FILTER
624			{
625			struct icmp_filter oval;
626			oval.data = 0xffffffff & ~(1 << ICMP4_ECHO_REPLY);
627			setsockopt (icmp4_fd, SOL_RAW, ICMP_FILTER, &oval, sizeof oval);
628			}
629			#endif
630
631	root	1.5	#if ENABLE_IPV6
632	root	1.1	icmp6_fd = socket (AF_INET6, SOCK_RAW, IPPROTO_ICMPV6);
633	root	1.2	fcntl (icmp6_fd, F_SETFL, O_NONBLOCK);
634	root	1.1	# ifdef ICMP6_FILTER
635			{
636			struct icmp6_filter oval;
637			ICMP6_FILTER_SETBLOCKALL (&oval);
638			ICMP6_FILTER_SETPASS (ICMP6_ECHO_REPLY, &oval);
639			setsockopt (icmp6_fd, IPPROTO_ICMPV6, ICMP6_FILTER, &oval, sizeof oval);
640			}
641			# endif
642			#endif
643
644			sv_setiv (get_sv ("AnyEvent::FastPing::ICMP4_FD", 1), icmp4_fd);
645			sv_setiv (get_sv ("AnyEvent::FastPing::ICMP6_FD", 1), icmp6_fd);
646			}
647
648	root	1.10	MODULE = AnyEvent::FastPing PACKAGE = AnyEvent::FastPing PREFIX = pinger_
649	root	1.1
650			BOOT:
651			{
652			HV *stash = gv_stashpv ("AnyEvent::FastPing", 1);
653
654	root	1.9	if (sizeof (PKT) & 3)
655			croak ("size of PKT structure is not a multiple of 4");
656
657	root	1.1	boot ();
658
659			newCONSTSUB (stash, "ipv4_supported", newSViv (icmp4_fd >= 0));
660			newCONSTSUB (stash, "ipv6_supported", newSViv (icmp6_fd >= 0));
661
662			newCONSTSUB (stash, "icmp4_pktsize", newSViv (HDR_SIZE_IP4 + sizeof (PKT)));
663			newCONSTSUB (stash, "icmp6_pktsize", newSViv (HDR_SIZE_IP6 + sizeof (PKT)));
664			}
665
666			PROTOTYPES: DISABLE
667
668	root	1.10	void
669	root	1.11	_recv_icmp4 (...)
670			CODE:
671			{
672			char buf [512];
673			struct sockaddr_in sa;
674			int maxrecv;
675
676			for (maxrecv = 256+1; --maxrecv; )
677			{
678			PINGER *pinger;
679			IP4HDR iphdr = (IP4HDR )buf;
680			socklen_t sl = sizeof (sa);
681			int len = recvfrom (icmp4_fd, buf, sizeof (buf), MSG_TRUNC, (struct sockaddr *)&sa, &sl);
682			int hdrlen, totlen;
683			PKT *pkt;
684
685			if (len <= HDR_SIZE_IP4)
686			break;
687
688			hdrlen = (iphdr->version_ihl & 15) * 4;
689			totlen = ntohs (iphdr->tot_len);
690
691			if (totlen > len
692			\|\| iphdr->protocol != IPPROTO_ICMP
693			\|\| hdrlen < HDR_SIZE_IP4 \|\| hdrlen + sizeof (PKT) != totlen)
694			continue;
695
696			pkt = (PKT *)(buf + hdrlen);
697
698			if (pkt->type != ICMP4_ECHO_REPLY
699			\|\| pkt->pinger >= pingercnt
700			\|\| !pingers [pkt->pinger])
701			continue;
702
703			pinger = pingers [pkt->pinger];
704
705			if (!pkt_is_valid_for (pkt, pinger))
706			continue;
707
708			recv_feed (pinger, &sa.sin_addr, 4, NOW () - pkt_to_ts (pkt));
709			}
710
711			recv_flush ();
712			}
713
714			void
715			_recv_icmp6 (...)
716			CODE:
717			{
718			struct sockaddr_in6 sa;
719			PKT pkt;
720			int maxrecv;
721
722			for (maxrecv = 256+1; --maxrecv; )
723			{
724			PINGER *pinger;
725			socklen_t sl = sizeof (sa);
726			int len = recvfrom (icmp6_fd, &pkt, sizeof (pkt), MSG_TRUNC, (struct sockaddr *)&sa, &sl);
727
728			if (len != sizeof (PKT))
729			break;
730
731			if (pkt.type != ICMP6_ECHO_REPLY
732			\|\| pkt.pinger >= pingercnt
733			\|\| !pingers [pkt.pinger])
734			continue;
735
736			pinger = pingers [pkt.pinger];
737
738			if (!pkt_is_valid_for (&pkt, pinger))
739			continue;
740
741			recv_feed (pinger, &sa.sin6_addr, 16, NOW () - pkt_to_ts (&pkt));
742			}
743
744			recv_flush ();
745			}
746
747			void
748	root	1.10	_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	root	1.11	XPUSHs (sv_2mortal (sv_bless (newRV_noinc (pv), gv_stashpv (SvPVutf8_nolen (klass), 1))));
755	root	1.10	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	root	1.1	CODE:
769	root	1.10	RETVAL = self->id;
770			OUTPUT:
771			RETVAL
772
773			void pinger_start (PINGER *self)
774
775			void pinger_stop (PINGER *self)
776	root	1.6
777	root	1.11	void
778			_stop_id (UV id)
779	root	1.10	CODE:
780			if (id < pingercnt && pingers [id])
781			pinger_stop (pingers [id]);
782	root	1.1
783	root	1.11	void
784			interval (PINGER *self, NV interval)
785	root	1.10	CODE:
786			self->interval = interval > MIN_INTERVAL ? interval : MIN_INTERVAL;
787	root	1.1
788	root	1.11	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	root	1.10	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	root	1.1
802	root	1.10	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	root	1.1
805	root	1.10	if (lo_len == 4 && icmp4_fd < 0) croak ("IPv4 support unavailable");
806			if (lo_len == 16 && icmp6_fd < 0) croak ("IPv6 support unavailable");
807	root	1.1
808	root	1.11	if (memcmp (lo, hi, lo_len) > 0)
809			croak ("AnyEvent::FastPing::add_range called with lo > hi");
810
811	root	1.10	range = calloc (1, sizeof (RANGE));
812	root	1.1
813	root	1.10	range->next = 0;
814			range->interval = interval > MIN_INTERVAL ? interval : MIN_INTERVAL;
815			range->addrlen = lo_len;
816	root	1.1
817	root	1.10	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	root	1.1
820	root	1.11	pinger_add_range (self, range);
821	root	1.1	}
822
823	root	1.10	void
824	root	1.11	add_hosts (PINGER self, SV addrs, NV interval)
825	root	1.1	CODE:
826	root	1.11	{
827			AV *av;
828			int i;
829			int cnt;
830			int addrlen;
831			RANGE *range;
832	root	1.1
833	root	1.11	if (!SvROK (addrs) \|\| SvTYPE (SvRV (addrs)) != SVt_PVAV)
834			croak ("AnyEvent::FastPing::add_hosst expects an arrayref with binary IPv4 or IPv6 addresses");
835	root	1.9
836	root	1.11	av = (AV *)SvRV (addrs);
837			cnt = av_len (av) + 1;
838	root	1.1
839	root	1.11	if (!cnt)
840			XSRETURN_EMPTY;
841	root	1.1
842	root	1.11	addrlen = SvCUR (*av_fetch (av, 0, 1));
843	root	1.1
844	root	1.11	if (addrlen != 4 && addrlen != 16)
845			croak ("AnyEvent::FastPing::add_hosts addresses must be specified as binary IPv4 or IPv6 addresses");
846	root	1.1
847	root	1.11	for (i = cnt; --i; )
848			{
849			SV sv = av_fetch (av, i, 1);
850	root	1.1
851	root	1.11	if (!sv_utf8_downgrade (sv, 1) \|\| addrlen != SvCUR (sv))
852			croak ("AnyEvent::FastPing::add_hosts addresses must all have the same size");
853			}
854	root	1.1
855	root	1.11	range = calloc (1, sizeof (RANGE) + cnt * addrlen);
856	root	1.10
857	root	1.11	range->next = 0;
858			range->interval = interval > MIN_INTERVAL ? interval : MIN_INTERVAL;
859			range->addrlen = addrlen;
860			range->addrcnt = cnt;
861	root	1.1
862	root	1.11	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	root	1.1
867	root	1.11	pinger_add_range (self, range);
868	root	1.1	}
869
870			void
871	root	1.11	on_recv (PINGER self, SV cb)
872	root	1.1	CODE:
873	root	1.11	SvREFCNT_dec (self->recvcb);
874			self->recvcb = newSVsv (cb);
875	root	1.1