[disorder] / clients / rtpmon.c

/*
 * This file is part of DisOrder.
 * Copyright (C) 2009 Richard Kettlewell
 *
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 * 
 * You should have received a copy of the GNU General Public License
 * along with this program.  If not, see <http://www.gnu.org/licenses/>.
 */
/** @file clients/rtpmon.c
 * @brief RTP monitor
 *
 * This progam monitors the rate at which data arrives by RTP and
 * constantly display it.  It is intended for debugging only.
 *
 * TODO de-dupe with playrtp.
 */
#include "common.h"

#include <getopt.h>
#include <sys/socket.h>
#include <sys/types.h>
#include <netdb.h>
#include <netinet/in.h>
#include <sys/time.h>
#include <unistd.h>
#include <locale.h>
#include <errno.h>
#include <stdlib.h>
#include <sys/uio.h>

#include "syscalls.h"
#include "timeval.h"
#include "mem.h"
#include "log.h"
#include "version.h"
#include "addr.h"
#include "configuration.h"
#include "rtp.h"

/** @brief Record of one packet */
struct entry {
  /** @brief When packet arrived */
  struct timeval when;

  /** @brief Serial number of first sample */
  uint32_t serial;
};

/** @brief Bytes per frame */
static unsigned bpf = 4;

/** @brief Frame serial number */
static uint32_t serial;

/** @brief Size of ring buffer */
#define RINGSIZE 131072

/** @brief Ring buffer */
static struct entry ring[RINGSIZE];

/** @brief Where new packets join the ring */
static unsigned ringtail;

static const struct option options[] = {
  { "help", no_argument, 0, 'h' },
  { "version", no_argument, 0, 'V' },
  { "bpf", required_argument, 0, 'b' },
  { 0, 0, 0, 0 }
};

static void help(void) {
  xprintf("Usage:\n"
	  "  rtpmon [OPTIONS] [ADDRESS] PORT\n"
	  "Options:\n"
          "  --bpf, -b               Bytes/frame (default 4)\n"
	  "  --help, -h              Display usage message\n"
	  "  --version, -V           Display version number\n"
          );
  xfclose(stdout);
  exit(0);
}

/** @brief Compute the rate by sampling at two points in the ring buffer */
static double rate(unsigned earlier, unsigned later) {
  const uint32_t frames = ring[later].serial - ring[earlier].serial;
  const int64_t us = tvsub_us(ring[later].when, ring[earlier].when);

  if(us)  
    return 1000000.0 * frames / us;
  else
    return 0.0;
}

/** @brief Called to say we received some bytes
 * @param when When we received them
 * @param n How many frames of audio data we received
 */
static void frames(const struct timeval *when, size_t n) {
  const time_t prev = ring[(ringtail - 1) % RINGSIZE].when.tv_sec;

  ring[ringtail].when = *when;
  ring[ringtail].serial = serial;
  serial += n;
  ringtail = (ringtail + 1) % RINGSIZE;
  // Report once a second
  if(prev != when->tv_sec) {
    if(printf("%8.2f  %8.2f  %8.2f  %8.2f  %8.2f  %8.2f  %8.2f\n",
              rate((ringtail - RINGSIZE / 128) % RINGSIZE,
                   (ringtail - 1) % RINGSIZE),
              rate((ringtail - RINGSIZE / 64) % RINGSIZE,
                   (ringtail - 1) % RINGSIZE),
              rate((ringtail - RINGSIZE / 32) % RINGSIZE,
                   (ringtail - 1) % RINGSIZE),
              rate((ringtail - RINGSIZE / 16) % RINGSIZE,
                   (ringtail - 1) % RINGSIZE),
              rate((ringtail - RINGSIZE / 8) % RINGSIZE,
                   (ringtail - 1) % RINGSIZE),
              rate((ringtail - RINGSIZE / 4) % RINGSIZE,
                   (ringtail - 1) % RINGSIZE),
              rate((ringtail - RINGSIZE / 2) % RINGSIZE,
                   (ringtail - 1) % RINGSIZE)) < 0
       || fflush(stdout) < 0)
      fatal(errno, "stdout");
  }
}

int main(int argc, char **argv) {
  int n;
  struct addrinfo *res;
  struct stringlist sl;
  struct ip_mreq mreq;
  struct ipv6_mreq mreq6;
  int rtpfd;
  char *sockname;
  int is_multicast;
  union any_sockaddr {
    struct sockaddr sa;
    struct sockaddr_in in;
    struct sockaddr_in6 in6;
  };
  union any_sockaddr mgroup;

  static const struct addrinfo prefs = {
    .ai_flags = AI_PASSIVE,
    .ai_family = PF_INET,
    .ai_socktype = SOCK_DGRAM,
    .ai_protocol = IPPROTO_UDP
  };
  static const int one = 1;

  mem_init();
  if(!setlocale(LC_CTYPE, "")) 
    fatal(errno, "error calling setlocale");
  while((n = getopt_long(argc, argv, "hVb:", options, 0)) >= 0) {
    switch(n) {
    case 'h': help();
    case 'V': version("rtpmon");
    case 'b': bpf = atoi(optarg); break;
    default: fatal(0, "invalid option");
    }
  }
  argc -= optind;
  argv += optind;
  switch(argc) {
  case 1:
  case 2:
    /* Use command-line ADDRESS+PORT or just PORT */
    sl.n = argc;
    sl.s = argv;
    break;
  default:
    fatal(0, "usage: rtpmon [OPTIONS] [ADDRESS] PORT");
  }
  if(!(res = get_address(&sl, &prefs, &sockname)))
    exit(1);
  /* Create the socket */
  if((rtpfd = socket(res->ai_family,
                     res->ai_socktype,
                     res->ai_protocol)) < 0)
    fatal(errno, "error creating socket");
  /* Allow multiple listeners */
  xsetsockopt(rtpfd, SOL_SOCKET, SO_REUSEADDR, &one, sizeof one);
  is_multicast = multicast(res->ai_addr);
  /* The multicast and unicast/broadcast cases are different enough that they
   * are totally split.  Trying to find commonality between them causes more
   * trouble that it's worth. */
  if(is_multicast) {
    /* Stash the multicast group address */
    memcpy(&mgroup, res->ai_addr, res->ai_addrlen);
    switch(res->ai_addr->sa_family) {
    case AF_INET:
      mgroup.in.sin_port = 0;
      break;
    case AF_INET6:
      mgroup.in6.sin6_port = 0;
      break;
    default:
      fatal(0, "unsupported family %d", (int)res->ai_addr->sa_family);
    }
    /* Bind to to the multicast group address */
    if(bind(rtpfd, res->ai_addr, res->ai_addrlen) < 0)
      fatal(errno, "error binding socket to %s", format_sockaddr(res->ai_addr));
    /* Add multicast group membership */
    switch(mgroup.sa.sa_family) {
    case PF_INET:
      mreq.imr_multiaddr = mgroup.in.sin_addr;
      mreq.imr_interface.s_addr = 0;      /* use primary interface */
      if(setsockopt(rtpfd, IPPROTO_IP, IP_ADD_MEMBERSHIP,
                    &mreq, sizeof mreq) < 0)
        fatal(errno, "error calling setsockopt IP_ADD_MEMBERSHIP");
      break;
    case PF_INET6:
      mreq6.ipv6mr_multiaddr = mgroup.in6.sin6_addr;
      memset(&mreq6.ipv6mr_interface, 0, sizeof mreq6.ipv6mr_interface);
      if(setsockopt(rtpfd, IPPROTO_IPV6, IPV6_JOIN_GROUP,
                    &mreq6, sizeof mreq6) < 0)
        fatal(errno, "error calling setsockopt IPV6_JOIN_GROUP");
      break;
    default:
      fatal(0, "unsupported address family %d", res->ai_family);
    }
    /* Report what we did */
    info("listening on %s multicast group %s",
         format_sockaddr(res->ai_addr), format_sockaddr(&mgroup.sa));
  } else {
    /* Bind to 0/port */
    switch(res->ai_addr->sa_family) {
    case AF_INET: {
      struct sockaddr_in *in = (struct sockaddr_in *)res->ai_addr;
      
      memset(&in->sin_addr, 0, sizeof (struct in_addr));
      if(bind(rtpfd, res->ai_addr, res->ai_addrlen) < 0)
        fatal(errno, "error binding socket to 0.0.0.0 port %d",
              ntohs(in->sin_port));
      break;
    }
    case AF_INET6: {
      struct sockaddr_in6 *in6 = (struct sockaddr_in6 *)res->ai_addr;
      
      memset(&in6->sin6_addr, 0, sizeof (struct in6_addr));
      break;
    }
    default:
      fatal(0, "unsupported family %d", (int)res->ai_addr->sa_family);
    }
    if(bind(rtpfd, res->ai_addr, res->ai_addrlen) < 0)
      fatal(errno, "error binding socket to %s", format_sockaddr(res->ai_addr));
    /* Report what we did */
    info("listening on %s", format_sockaddr(res->ai_addr));
  }
  for(;;) {
    struct rtp_header header;
    char buffer[4096];
    struct iovec iov[2];
    struct timeval when;

    iov[0].iov_base = &header;
    iov[0].iov_len = sizeof header;
    iov[1].iov_base = buffer;
    iov[1].iov_len = sizeof buffer;
    n = readv(rtpfd, iov, 2);
    gettimeofday(&when, 0);
    if(n < 0) {
      switch(errno) {
      case EINTR:
        continue;
      default:
        fatal(errno, "error reading from socket");
      }
    }
    if((size_t)n <= sizeof (struct rtp_header)) {
      info("ignored a short packet");
      continue;
    }
    frames(&when, (n - sizeof header) / bpf);
  }
}

/*
Local Variables:
c-basic-offset:2
comment-column:40
fill-column:79
indent-tabs-mode:nil
End:
*/
Commit	Line	Data
210d172f RK	1	/*
	2	* This file is part of DisOrder.
	3	* Copyright (C) 2009 Richard Kettlewell
	4	*
	5	* This program is free software: you can redistribute it and/or modify
	6	* it under the terms of the GNU General Public License as published by
	7	* the Free Software Foundation, either version 3 of the License, or
	8	* (at your option) any later version.
	9	*
	10	* This program is distributed in the hope that it will be useful,
	11	* but WITHOUT ANY WARRANTY; without even the implied warranty of
	12	* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
	13	* GNU General Public License for more details.
	14	*
	15	* You should have received a copy of the GNU General Public License
	16	* along with this program. If not, see <http://www.gnu.org/licenses/>.
	17	*/
	18	/** @file clients/rtpmon.c
	19	* @brief RTP monitor
	20	*
	21	* This progam monitors the rate at which data arrives by RTP and
	22	* constantly display it. It is intended for debugging only.
	23	*
	24	* TODO de-dupe with playrtp.
	25	*/
	26	#include "common.h"
	27
	28	#include <getopt.h>
	29	#include <sys/socket.h>
	30	#include <sys/types.h>
	31	#include <netdb.h>
	32	#include <netinet/in.h>
	33	#include <sys/time.h>
	34	#include <unistd.h>
	35	#include <locale.h>
	36	#include <errno.h>
	37	#include <stdlib.h>
033ede99	38	#include <sys/uio.h>
210d172f RK	39
	40	#include "syscalls.h"
	41	#include "timeval.h"
	42	#include "mem.h"
	43	#include "log.h"
	44	#include "version.h"
	45	#include "addr.h"
	46	#include "configuration.h"
	47	#include "rtp.h"
	48
	49	/** @brief Record of one packet */
	50	struct entry {
	51	/** @brief When packet arrived */
	52	struct timeval when;
	53
	54	/** @brief Serial number of first sample */
	55	uint32_t serial;
	56	};
	57
	58	/** @brief Bytes per frame */
	59	static unsigned bpf = 4;
	60
	61	/** @brief Frame serial number */
	62	static uint32_t serial;
	63
	64	/** @brief Size of ring buffer */
211af11e	65	#define RINGSIZE 131072
210d172f RK	66
	67	/** @brief Ring buffer */
	68	static struct entry ring[RINGSIZE];
	69
	70	/** @brief Where new packets join the ring */
	71	static unsigned ringtail;
	72
	73	static const struct option options[] = {
	74	{ "help", no_argument, 0, 'h' },
	75	{ "version", no_argument, 0, 'V' },
	76	{ "bpf", required_argument, 0, 'b' },
	77	{ 0, 0, 0, 0 }
	78	};
	79
	80	static void help(void) {
	81	xprintf("Usage:\n"
	82	" rtpmon [OPTIONS] [ADDRESS] PORT\n"
	83	"Options:\n"
	84	" --bpf, -b Bytes/frame (default 4)\n"
	85	" --help, -h Display usage message\n"
	86	" --version, -V Display version number\n"
	87	);
	88	xfclose(stdout);
	89	exit(0);
	90	}
	91
	92	/** @brief Compute the rate by sampling at two points in the ring buffer */
	93	static double rate(unsigned earlier, unsigned later) {
	94	const uint32_t frames = ring[later].serial - ring[earlier].serial;
	95	const int64_t us = tvsub_us(ring[later].when, ring[earlier].when);
	96
	97	if(us)
	98	return 1000000.0 * frames / us;
	99	else
	100	return 0.0;
	101	}
	102
	103	/** @brief Called to say we received some bytes
	104	* @param when When we received them
	105	* @param n How many frames of audio data we received
	106	*/
	107	static void frames(const struct timeval *when, size_t n) {
92dcf839 RK	108	const time_t prev = ring[(ringtail - 1) % RINGSIZE].when.tv_sec;
92dcf839 RK	109
210d172f RK	110	ring[ringtail].when = *when;
	111	ring[ringtail].serial = serial;
	112	serial += n;
	113	ringtail = (ringtail + 1) % RINGSIZE;
92dcf839 RK	114	// Report once a second
92dcf839 RK	115	if(prev != when->tv_sec) {
211af11e RK	116	if(printf("%8.2f %8.2f %8.2f %8.2f %8.2f %8.2f %8.2f\n",
	117	rate((ringtail - RINGSIZE / 128) % RINGSIZE,
	118	(ringtail - 1) % RINGSIZE),
	119	rate((ringtail - RINGSIZE / 64) % RINGSIZE,
	120	(ringtail - 1) % RINGSIZE),
	121	rate((ringtail - RINGSIZE / 32) % RINGSIZE,
	122	(ringtail - 1) % RINGSIZE),
	123	rate((ringtail - RINGSIZE / 16) % RINGSIZE,
	124	(ringtail - 1) % RINGSIZE),
210d172f RK	125	rate((ringtail - RINGSIZE / 8) % RINGSIZE,
	126	(ringtail - 1) % RINGSIZE),
	127	rate((ringtail - RINGSIZE / 4) % RINGSIZE,
	128	(ringtail - 1) % RINGSIZE),
	129	rate((ringtail - RINGSIZE / 2) % RINGSIZE,
	130	(ringtail - 1) % RINGSIZE)) < 0
	131	\|\| fflush(stdout) < 0)
	132	fatal(errno, "stdout");
	133	}
	134	}
	135
	136	int main(int argc, char **argv) {
	137	int n;
	138	struct addrinfo *res;
	139	struct stringlist sl;
	140	struct ip_mreq mreq;
	141	struct ipv6_mreq mreq6;
	142	int rtpfd;
	143	char *sockname;
	144	int is_multicast;
	145	union any_sockaddr {
	146	struct sockaddr sa;
	147	struct sockaddr_in in;
	148	struct sockaddr_in6 in6;
	149	};
	150	union any_sockaddr mgroup;
	151
	152	static const struct addrinfo prefs = {
	153	.ai_flags = AI_PASSIVE,
	154	.ai_family = PF_INET,
	155	.ai_socktype = SOCK_DGRAM,
	156	.ai_protocol = IPPROTO_UDP
	157	};
	158	static const int one = 1;
	159
	160	mem_init();
	161	if(!setlocale(LC_CTYPE, ""))
	162	fatal(errno, "error calling setlocale");
	163	while((n = getopt_long(argc, argv, "hVb:", options, 0)) >= 0) {
	164	switch(n) {
	165	case 'h': help();
	166	case 'V': version("rtpmon");
	167	case 'b': bpf = atoi(optarg); break;
	168	default: fatal(0, "invalid option");
	169	}
	170	}
	171	argc -= optind;
	172	argv += optind;
	173	switch(argc) {
	174	case 1:
	175	case 2:
	176	/* Use command-line ADDRESS+PORT or just PORT */
	177	sl.n = argc;
	178	sl.s = argv;
	179	break;
	180	default:
	181	fatal(0, "usage: rtpmon [OPTIONS] [ADDRESS] PORT");
	182	}
	183	if(!(res = get_address(&sl, &prefs, &sockname)))
	184	exit(1);
	185	/* Create the socket */
	186	if((rtpfd = socket(res->ai_family,
	187	res->ai_socktype,
	188	res->ai_protocol)) < 0)
189	fatal(errno, "error creating socket");
190	/* Allow multiple listeners */
191	xsetsockopt(rtpfd, SOL_SOCKET, SO_REUSEADDR, &one, sizeof one);
192	is_multicast = multicast(res->ai_addr);
193	/* The multicast and unicast/broadcast cases are different enough that they
194	* are totally split. Trying to find commonality between them causes more
195	* trouble that it's worth. */
196	if(is_multicast) {
197	/* Stash the multicast group address */
198	memcpy(&mgroup, res->ai_addr, res->ai_addrlen);
199	switch(res->ai_addr->sa_family) {
200	case AF_INET:
201	mgroup.in.sin_port = 0;
202	break;
203	case AF_INET6:
204	mgroup.in6.sin6_port = 0;
205	break;
206	default:
207	fatal(0, "unsupported family %d", (int)res->ai_addr->sa_family);
208	}
209	/* Bind to to the multicast group address */
210	if(bind(rtpfd, res->ai_addr, res->ai_addrlen) < 0)
211	fatal(errno, "error binding socket to %s", format_sockaddr(res->ai_addr));
212	/* Add multicast group membership */
213	switch(mgroup.sa.sa_family) {
214	case PF_INET:
215	mreq.imr_multiaddr = mgroup.in.sin_addr;
216	mreq.imr_interface.s_addr = 0; /* use primary interface */
217	if(setsockopt(rtpfd, IPPROTO_IP, IP_ADD_MEMBERSHIP,
218	&mreq, sizeof mreq) < 0)
219	fatal(errno, "error calling setsockopt IP_ADD_MEMBERSHIP");
220	break;
221	case PF_INET6:
222	mreq6.ipv6mr_multiaddr = mgroup.in6.sin6_addr;
223	memset(&mreq6.ipv6mr_interface, 0, sizeof mreq6.ipv6mr_interface);
224	if(setsockopt(rtpfd, IPPROTO_IPV6, IPV6_JOIN_GROUP,
225	&mreq6, sizeof mreq6) < 0)
226	fatal(errno, "error calling setsockopt IPV6_JOIN_GROUP");
227	break;
228	default:
229	fatal(0, "unsupported address family %d", res->ai_family);
230	}
231	/* Report what we did */
232	info("listening on %s multicast group %s",
233	format_sockaddr(res->ai_addr), format_sockaddr(&mgroup.sa));
234	} else {
235	/* Bind to 0/port */
236	switch(res->ai_addr->sa_family) {
237	case AF_INET: {
238	struct sockaddr_in in = (struct sockaddr_in )res->ai_addr;
239
240	memset(&in->sin_addr, 0, sizeof (struct in_addr));
241	if(bind(rtpfd, res->ai_addr, res->ai_addrlen) < 0)
242	fatal(errno, "error binding socket to 0.0.0.0 port %d",
243	ntohs(in->sin_port));
244	break;
245	}
246	case AF_INET6: {
247	struct sockaddr_in6 in6 = (struct sockaddr_in6 )res->ai_addr;
248
249	memset(&in6->sin6_addr, 0, sizeof (struct in6_addr));
250	break;
251	}
252	default:
253	fatal(0, "unsupported family %d", (int)res->ai_addr->sa_family);
254	}
255	if(bind(rtpfd, res->ai_addr, res->ai_addrlen) < 0)
256	fatal(errno, "error binding socket to %s", format_sockaddr(res->ai_addr));
257	/* Report what we did */
258	info("listening on %s", format_sockaddr(res->ai_addr));
259	}
260	for(;;) {
261	struct rtp_header header;
262	char buffer[4096];
263	struct iovec iov[2];
264	struct timeval when;
265
266	iov[0].iov_base = &header;
267	iov[0].iov_len = sizeof header;
268	iov[1].iov_base = buffer;
269	iov[1].iov_len = sizeof buffer;
270	n = readv(rtpfd, iov, 2);
271	gettimeofday(&when, 0);
272	if(n < 0) {
273	switch(errno) {
274	case EINTR:
275	continue;
276	default:
277	fatal(errno, "error reading from socket");
278	}
279	}
280	if((size_t)n <= sizeof (struct rtp_header)) {
281	info("ignored a short packet");
282	continue;
283	}
284	frames(&when, (n - sizeof header) / bpf);
285	}
286	}
287
288	/*
289	Local Variables:
290	c-basic-offset:2
291	comment-column:40
292	fill-column:79
293	indent-tabs-mode:nil
294	End:
295	*/