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