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