chiark - git - mdw - disorder/blob - clients/rtpmon.c

   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) {
 107   const time_t prev = ring[(ringtail - 1) % RINGSIZE].when.tv_sec;
 108
 109   ring[ringtail].when = *when;
 110   ring[ringtail].serial = serial;
 111   serial += n;
 112   ringtail = (ringtail + 1) % RINGSIZE;
 113   // Report once a second
 114   if(prev != when->tv_sec) {
 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 */