[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 16384

/** @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\n",
              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 */
	64	#define RINGSIZE 16384
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) {
210d172f RK	115	if(printf("%8.2f %8.2f %8.2f\n",
	116	rate((ringtail - RINGSIZE / 8) % RINGSIZE,
	117	(ringtail - 1) % RINGSIZE),
	118	rate((ringtail - RINGSIZE / 4) % RINGSIZE,
	119	(ringtail - 1) % RINGSIZE),
	120	rate((ringtail - RINGSIZE / 2) % RINGSIZE,
	121	(ringtail - 1) % RINGSIZE)) < 0
	122	\|\| fflush(stdout) < 0)
	123	fatal(errno, "stdout");
	124	}
	125	}
	126
	127	int main(int argc, char **argv) {
	128	int n;
	129	struct addrinfo *res;
	130	struct stringlist sl;
	131	struct ip_mreq mreq;
	132	struct ipv6_mreq mreq6;
	133	int rtpfd;
	134	char *sockname;
	135	int is_multicast;
	136	union any_sockaddr {
	137	struct sockaddr sa;
	138	struct sockaddr_in in;
	139	struct sockaddr_in6 in6;
	140	};
	141	union any_sockaddr mgroup;
	142
	143	static const struct addrinfo prefs = {
	144	.ai_flags = AI_PASSIVE,
	145	.ai_family = PF_INET,
	146	.ai_socktype = SOCK_DGRAM,
	147	.ai_protocol = IPPROTO_UDP
	148	};
	149	static const int one = 1;
	150
	151	mem_init();
	152	if(!setlocale(LC_CTYPE, ""))
	153	fatal(errno, "error calling setlocale");
	154	while((n = getopt_long(argc, argv, "hVb:", options, 0)) >= 0) {
	155	switch(n) {
	156	case 'h': help();
	157	case 'V': version("rtpmon");
	158	case 'b': bpf = atoi(optarg); break;
	159	default: fatal(0, "invalid option");
	160	}
	161	}
	162	argc -= optind;
	163	argv += optind;
	164	switch(argc) {
	165	case 1:
	166	case 2:
	167	/* Use command-line ADDRESS+PORT or just PORT */
	168	sl.n = argc;
	169	sl.s = argv;
	170	break;
	171	default:
	172	fatal(0, "usage: rtpmon [OPTIONS] [ADDRESS] PORT");
	173	}
	174	if(!(res = get_address(&sl, &prefs, &sockname)))
	175	exit(1);
	176	/* Create the socket */
	177	if((rtpfd = socket(res->ai_family,
	178	res->ai_socktype,
179	res->ai_protocol)) < 0)
180	fatal(errno, "error creating socket");
181	/* Allow multiple listeners */
182	xsetsockopt(rtpfd, SOL_SOCKET, SO_REUSEADDR, &one, sizeof one);
183	is_multicast = multicast(res->ai_addr);
184	/* The multicast and unicast/broadcast cases are different enough that they
185	* are totally split. Trying to find commonality between them causes more
186	* trouble that it's worth. */
187	if(is_multicast) {
188	/* Stash the multicast group address */
189	memcpy(&mgroup, res->ai_addr, res->ai_addrlen);
190	switch(res->ai_addr->sa_family) {
191	case AF_INET:
192	mgroup.in.sin_port = 0;
193	break;
194	case AF_INET6:
195	mgroup.in6.sin6_port = 0;
196	break;
197	default:
198	fatal(0, "unsupported family %d", (int)res->ai_addr->sa_family);
199	}
200	/* Bind to to the multicast group address */
201	if(bind(rtpfd, res->ai_addr, res->ai_addrlen) < 0)
202	fatal(errno, "error binding socket to %s", format_sockaddr(res->ai_addr));
203	/* Add multicast group membership */
204	switch(mgroup.sa.sa_family) {
205	case PF_INET:
206	mreq.imr_multiaddr = mgroup.in.sin_addr;
207	mreq.imr_interface.s_addr = 0; /* use primary interface */
208	if(setsockopt(rtpfd, IPPROTO_IP, IP_ADD_MEMBERSHIP,
209	&mreq, sizeof mreq) < 0)
210	fatal(errno, "error calling setsockopt IP_ADD_MEMBERSHIP");
211	break;
212	case PF_INET6:
213	mreq6.ipv6mr_multiaddr = mgroup.in6.sin6_addr;
214	memset(&mreq6.ipv6mr_interface, 0, sizeof mreq6.ipv6mr_interface);
215	if(setsockopt(rtpfd, IPPROTO_IPV6, IPV6_JOIN_GROUP,
216	&mreq6, sizeof mreq6) < 0)
217	fatal(errno, "error calling setsockopt IPV6_JOIN_GROUP");
218	break;
219	default:
220	fatal(0, "unsupported address family %d", res->ai_family);
221	}
222	/* Report what we did */
223	info("listening on %s multicast group %s",
224	format_sockaddr(res->ai_addr), format_sockaddr(&mgroup.sa));
225	} else {
226	/* Bind to 0/port */
227	switch(res->ai_addr->sa_family) {
228	case AF_INET: {
229	struct sockaddr_in in = (struct sockaddr_in )res->ai_addr;
230
231	memset(&in->sin_addr, 0, sizeof (struct in_addr));
232	if(bind(rtpfd, res->ai_addr, res->ai_addrlen) < 0)
233	fatal(errno, "error binding socket to 0.0.0.0 port %d",
234	ntohs(in->sin_port));
235	break;
236	}
237	case AF_INET6: {
238	struct sockaddr_in6 in6 = (struct sockaddr_in6 )res->ai_addr;
239
240	memset(&in6->sin6_addr, 0, sizeof (struct in6_addr));
241	break;
242	}
243	default:
244	fatal(0, "unsupported family %d", (int)res->ai_addr->sa_family);
245	}
246	if(bind(rtpfd, res->ai_addr, res->ai_addrlen) < 0)
247	fatal(errno, "error binding socket to %s", format_sockaddr(res->ai_addr));
248	/* Report what we did */
249	info("listening on %s", format_sockaddr(res->ai_addr));
250	}
251	for(;;) {
252	struct rtp_header header;
253	char buffer[4096];
254	struct iovec iov[2];
255	struct timeval when;
256
257	iov[0].iov_base = &header;
258	iov[0].iov_len = sizeof header;
259	iov[1].iov_base = buffer;
260	iov[1].iov_len = sizeof buffer;
261	n = readv(rtpfd, iov, 2);
262	gettimeofday(&when, 0);
263	if(n < 0) {
264	switch(errno) {
265	case EINTR:
266	continue;
267	default:
268	fatal(errno, "error reading from socket");
269	}
270	}
271	if((size_t)n <= sizeof (struct rtp_header)) {
272	info("ignored a short packet");
273	continue;
274	}
275	frames(&when, (n - sizeof header) / bpf);
276	}
277	}
278
279	/*
280	Local Variables:
281	c-basic-offset:2
282	comment-column:40
283	fill-column:79
284	indent-tabs-mode:nil
285	End:
286	*/