modules/access/rtp/session.c

   1 /**
   2  * @file session.c
   3  * @brief RTP session handling
   4  */
   5 /*****************************************************************************
   6  * Copyright © 2008 Rémi Denis-Courmont
   7  *
   8  * This library is free software; you can redistribute it and/or
   9  * modify it under the terms of the GNU Lesser General Public License
  10  * as published by the Free Software Foundation; either version 2.1
  11  * of the License, or (at your option) any later version.
  12  *
  13  * This library is distributed in the hope that it will be useful,
  14  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  15  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  16  * GNU Lesser General Public License for more details.
  17  *
  18  * You should have received a copy of the GNU Lesser General Public
  19  * License along with this library; if not, write to the Free Software
  20  * Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301  USA
  21  ****************************************************************************/
  22
  23 #ifdef HAVE_CONFIG_H
  24 # include <config.h>
  25 #endif
  26
  27 #include <stdlib.h>
  28 #include <assert.h>
  29 #include <errno.h>
  30
  31 #include <vlc_common.h>
  32 #include <vlc_demux.h>
  33
  34 #include "rtp.h"
  35
  36 typedef struct rtp_source_t rtp_source_t;
  37
  38 /** State for a RTP session: */
  39 struct rtp_session_t
  40 {
  41     rtp_source_t **srcv;
  42     unsigned       srcc;
  43     uint8_t        ptc;
  44     rtp_pt_t      *ptv;
  45 };
  46
  47 static rtp_source_t *
  48 rtp_source_create (demux_t *, const rtp_session_t *, uint32_t, uint16_t);
  49 static void
  50 rtp_source_destroy (demux_t *, const rtp_session_t *, rtp_source_t *);
  51
  52 static void rtp_decode (demux_t *, const rtp_session_t *, rtp_source_t *);
  53
  54 /**
  55  * Creates a new RTP session.
  56  */
  57 rtp_session_t *
  58 rtp_session_create (demux_t *demux)
  59 {
  60     rtp_session_t *session = malloc (sizeof (*session));
  61     if (session == NULL)
  62         return NULL;
  63
  64     session->srcv = NULL;
  65     session->srcc = 0;
  66     session->ptc = 0;
  67     session->ptv = NULL;
  68
  69     (void)demux;
  70     return session;
  71 }
  72
  73
  74 /**
  75  * Destroys an RTP session.
  76  */
  77 void rtp_session_destroy (demux_t *demux, rtp_session_t *session)
  78 {
  79     for (unsigned i = 0; i < session->srcc; i++)
  80         rtp_source_destroy (demux, session, session->srcv[i]);
  81
  82     free (session->srcv);
  83     free (session->ptv);
  84     free (session);
  85     (void)demux;
  86 }
  87
  88 static void *no_init (demux_t *demux)
  89 {
  90     (void)demux;
  91     return NULL;
  92 }
  93
  94 static void no_destroy (demux_t *demux, void *opaque)
  95 {
  96     (void)demux; (void)opaque;
  97 }
  98
  99 static void no_decode (demux_t *demux, void *opaque, block_t *block)
 100 {
 101     (void)demux; (void)opaque;
 102     block_Release (block);
 103 }
 104
 105 /**
 106  * Adds a payload type to an RTP session.
 107  */
 108 int rtp_add_type (demux_t *demux, rtp_session_t *ses, const rtp_pt_t *pt)
 109 {
 110     if (ses->srcc > 0)
 111     {
 112         msg_Err (demux, "cannot change RTP payload formats during session");
 113         return EINVAL;
 114     }
 115
 116     rtp_pt_t *ppt = realloc (ses->ptv, (ses->ptc + 1) * sizeof (rtp_pt_t));
 117     if (ppt == NULL)
 118         return ENOMEM;
 119
 120     ses->ptv = ppt;
 121     ppt += ses->ptc++;
 122
 123     ppt->init = pt->init ? pt->init : no_init;
 124     ppt->destroy = pt->destroy ? pt->destroy : no_destroy;
 125     ppt->decode = pt->decode ? pt->decode : no_decode;
 126     ppt->header = NULL;
 127     ppt->frequency = pt->frequency;
 128     ppt->number = pt->number;
 129     msg_Dbg (demux, "added payload type %"PRIu8" (f = %"PRIu32" Hz)",
 130              ppt->number, ppt->frequency);
 131
 132     assert (ppt->frequency > 0); /* SIGFPE! */
 133     (void)demux;
 134     return 0;
 135 }
 136
 137 /** State for an RTP source */
 138 struct rtp_source_t
 139 {
 140     uint32_t ssrc;
 141     uint32_t jitter;  /* interarrival delay jitter estimate */
 142     mtime_t  last_rx; /* last received packet local timestamp */
 143     uint32_t last_ts; /* last received packet RTP timestamp */
 144
 145     uint32_t ref_rtp; /* sender RTP timestamp reference */
 146     mtime_t  ref_ntp; /* sender NTP timestamp reference */
 147
 148     uint16_t bad_seq; /* tentatively next expected sequence for resync */
 149     uint16_t max_seq; /* next expected sequence */
 150
 151     uint16_t last_seq; /* sequence of the next dequeued packet */
 152     block_t *blocks; /* re-ordered blocks queue */
 153     void    *opaque[]; /* Per-source private payload data */
 154 };
 155
 156 /**
 157  * Initializes a new RTP source within an RTP session.
 158  */
 159 static rtp_source_t *
 160 rtp_source_create (demux_t *demux, const rtp_session_t *session,
 161                    uint32_t ssrc, uint16_t init_seq)
 162 {
 163     rtp_source_t *source;
 164
 165     source = malloc (sizeof (*source) + (sizeof (void *) * session->ptc));
 166     if (source == NULL)
 167         return NULL;
 168
 169     source->ssrc = ssrc;
 170     source->jitter = 0;
 171     source->ref_rtp = 0;
 172     /* TODO: use VLC_TS_0, but VLC does not like negative PTS at the moment */
 173     source->ref_ntp = UINT64_C (1) << 62;
 174     source->max_seq = source->bad_seq = init_seq;
 175     source->last_seq = init_seq - 1;
 176     source->blocks = NULL;
 177
 178     /* Initializes all payload */
 179     for (unsigned i = 0; i < session->ptc; i++)
 180         source->opaque[i] = session->ptv[i].init (demux);
 181
 182     msg_Dbg (demux, "added RTP source (%08x)", ssrc);
 183     return source;
 184 }
 185
 186
 187 /**
 188  * Destroys an RTP source and its associated streams.
 189  */
 190 static void
 191 rtp_source_destroy (demux_t *demux, const rtp_session_t *session,
 192                     rtp_source_t *source)
 193 {
 194     msg_Dbg (demux, "removing RTP source (%08x)", source->ssrc);
 195
 196     for (unsigned i = 0; i < session->ptc; i++)
 197         session->ptv[i].destroy (demux, source->opaque[i]);
 198     block_ChainRelease (source->blocks);
 199     free (source);
 200 }
 201
 202 static inline uint16_t rtp_seq (const block_t *block)
 203 {
 204     assert (block->i_buffer >= 4);
 205     return GetWBE (block->p_buffer + 2);
 206 }
 207
 208 static inline uint32_t rtp_timestamp (const block_t *block)
 209 {
 210     assert (block->i_buffer >= 12);
 211     return GetDWBE (block->p_buffer + 4);
 212 }
 213
 214 static const struct rtp_pt_t *
 215 rtp_find_ptype (const rtp_session_t *session, rtp_source_t *source,
 216                 const block_t *block, void **pt_data)
 217 {
 218     uint8_t ptype = rtp_ptype (block);
 219
 220     for (unsigned i = 0; i < session->ptc; i++)
 221     {
 222         if (session->ptv[i].number == ptype)
 223         {
 224             if (pt_data != NULL)
 225                 *pt_data = source->opaque[i];
 226             return &session->ptv[i];
 227         }
 228     }
 229     return NULL;
 230 }
 231
 232 /**
 233  * Receives an RTP packet and queues it. Not a cancellation point.
 234  *
 235  * @param demux VLC demux object
 236  * @param session RTP session receiving the packet
 237  * @param block RTP packet including the RTP header
 238  */
 239 void
 240 rtp_queue (demux_t *demux, rtp_session_t *session, block_t *block)
 241 {
 242     demux_sys_t *p_sys = demux->p_sys;
 243
 244     /* RTP header sanity checks (see RFC 3550) */
 245     if (block->i_buffer < 12)
 246         goto drop;
 247     if ((block->p_buffer[0] >> 6 ) != 2) /* RTP version number */
 248         goto drop;
 249
 250     /* Remove padding if present */
 251     if (block->p_buffer[0] & 0x20)
 252     {
 253         uint8_t padding = block->p_buffer[block->i_buffer - 1];
 254         if ((padding == 0) || (block->i_buffer < (12u + padding)))
 255             goto drop; /* illegal value */
 256
 257         block->i_buffer -= padding;
 258     }
 259
 260     mtime_t        now = mdate ();
 261     rtp_source_t  *src  = NULL;
 262     const uint16_t seq  = rtp_seq (block);
 263     const uint32_t ssrc = GetDWBE (block->p_buffer + 8);
 264
 265     /* In most case, we know this source already */
 266     for (unsigned i = 0, max = session->srcc; i < max; i++)
 267     {
 268         rtp_source_t *tmp = session->srcv[i];
 269         if (tmp->ssrc == ssrc)
 270         {
 271             src = tmp;
 272             break;
 273         }
 274
 275         /* RTP source garbage collection */
 276         if ((tmp->last_rx + p_sys->timeout) < now)
 277         {
 278             rtp_source_destroy (demux, session, tmp);
 279             if (--session->srcc > 0)
 280                 session->srcv[i] = session->srcv[session->srcc - 1];
 281         }
 282     }
 283
 284     if (src == NULL)
 285     {
 286         /* New source */
 287         if (session->srcc >= p_sys->max_src)
 288         {
 289             msg_Warn (demux, "too many RTP sessions");
 290             goto drop;
 291         }
 292
 293         rtp_source_t **tab;
 294         tab = realloc (session->srcv, (session->srcc + 1) * sizeof (*tab));
 295         if (tab == NULL)
 296             goto drop;
 297         session->srcv = tab;
 298
 299         src = rtp_source_create (demux, session, ssrc, seq);
 300         if (src == NULL)
 301             goto drop;
 302
 303         tab[session->srcc++] = src;
 304         /* Cannot compute jitter yet */
 305     }
 306     else
 307     {
 308         const rtp_pt_t *pt = rtp_find_ptype (session, src, block, NULL);
 309
 310         if (pt != NULL)
 311         {
 312             /* Recompute jitter estimate.
 313              * That is computed from the RTP timestamps and the system clock.
 314              * It is independent of RTP sequence. */
 315             uint32_t freq = pt->frequency;
 316             int64_t ts = rtp_timestamp (block);
 317             int64_t d = ((now - src->last_rx) * freq) / CLOCK_FREQ;
 318             d        -=    ts - src->last_ts;
 319             if (d < 0) d = -d;
 320             src->jitter += ((d - src->jitter) + 8) >> 4;
 321         }
 322     }
 323     src->last_rx = now;
 324     block->i_pts = now; /* store reception time until dequeued */
 325     src->last_ts = rtp_timestamp (block);
 326
 327     /* Check sequence number */
 328     /* NOTE: the sequence number is per-source,
 329      * but is independent from the payload type. */
 330     int16_t delta_seq = seq - src->max_seq;
 331     if ((delta_seq > 0) ? (delta_seq > p_sys->max_dropout)
 332                         : (-delta_seq > p_sys->max_misorder))
 333     {
 334         msg_Dbg (demux, "sequence discontinuity"
 335                  " (got: %"PRIu16", expected: %"PRIu16")", seq, src->max_seq);
 336         if (seq == src->bad_seq)
 337         {
 338             src->max_seq = src->bad_seq = seq + 1;
 339             src->last_seq = seq - 0x7fffe; /* hack for rtp_decode() */
 340             msg_Warn (demux, "sequence resynchronized");
 341             block_ChainRelease (src->blocks);
 342             src->blocks = NULL;
 343         }
 344         else
 345         {
 346             src->bad_seq = seq + 1;
 347             goto drop;
 348         }
 349     }
 350     else
 351     if (delta_seq >= 0)
 352         src->max_seq = seq + 1;
 353
 354     /* Queues the block in sequence order,
 355      * hence there is a single queue for all payload types. */
 356     block_t **pp = &src->blocks;
 357     for (block_t *prev = *pp; prev != NULL; prev = *pp)
 358     {
 359         delta_seq = seq - rtp_seq (prev);
 360         if (delta_seq < 0)
 361             break;
 362         if (delta_seq == 0)
 363         {
 364             msg_Dbg (demux, "duplicate packet (sequence: %"PRIu16")", seq);
 365             goto drop; /* duplicate */
 366         }
 367         pp = &prev->p_next;
 368     }
 369     block->p_next = *pp;
 370     *pp = block;
 371
 372     /*rtp_decode (demux, session, src);*/
 373     return;
 374
 375 drop:
 376     block_Release (block);
 377 }
 378
 379
 380 static void rtp_decode (demux_t *, const rtp_session_t *, rtp_source_t *);
 381
 382 /**
 383  * Dequeues RTP packets and pass them to decoder. Not cancellation-safe(?).
 384  * A packet is decoded if it is the next in sequence order, or if we have
 385  * given up waiting on the missing packets (time out) from the last one
 386  * already decoded.
 387  *
 388  * @param demux VLC demux object
 389  * @param session RTP session receiving the packet
 390  * @param deadlinep pointer to deadline to call rtp_dequeue() again
 391  * @return true if the buffer is not empty, false otherwise.
 392  * In the later case, *deadlinep is undefined.
 393  */
 394 bool rtp_dequeue (demux_t *demux, const rtp_session_t *session,
 395                   mtime_t *restrict deadlinep)
 396 {
 397     mtime_t now = mdate ();
 398     bool pending = false;
 399
 400     *deadlinep = INT64_MAX;
 401
 402     for (unsigned i = 0, max = session->srcc; i < max; i++)
 403     {
 404         rtp_source_t *src = session->srcv[i];
 405         block_t *block;
 406
 407         /* Because of IP packet delay variation (IPDV), we need to guesstimate
 408          * how long to wait for a missing packet in the RTP sequence
 409          * (see RFC3393 for background on IPDV).
 410          *
 411          * This situation occurs if a packet got lost, or if the network has
 412          * re-ordered packets. Unfortunately, the MSL is 2 minutes, orders of
 413          * magnitude too long for multimedia. We need a trade-off.
 414          * If we underestimated IPDV, we may have to discard valid but late
 415          * packets. If we overestimate it, we will either cause too much
 416          * delay, or worse, underflow our downstream buffers, as we wait for
 417          * definitely a lost packets.
 418          *
 419          * The rest of the "de-jitter buffer" work is done by the internal
 420          * LibVLC E/S-out clock synchronization. Here, we need to bother about
 421          * re-ordering packets, as decoders can't cope with mis-ordered data.
 422          */
 423         while (((block = src->blocks)) != NULL)
 424         {
 425             if ((int16_t)(rtp_seq (block) - (src->last_seq + 1)) <= 0)
 426             {   /* Next (or earlier) block ready, no need to wait */
 427                 rtp_decode (demux, session, src);
 428                 continue;
 429             }
 430
 431             /* Wait for 3 times the inter-arrival delay variance (about 99.7%
 432              * match for random gaussian jitter).
 433              */
 434             mtime_t deadline;
 435             const rtp_pt_t *pt = rtp_find_ptype (session, src, block, NULL);
 436             if (pt)
 437                 deadline = CLOCK_FREQ * 3 * src->jitter / pt->frequency;
 438             else
 439                 deadline = 0; /* no jitter estimate with no frequency :( */
 440
 441             /* Make sure we wait at least for 25 msec */
 442             if (deadline < (CLOCK_FREQ / 40))
 443                 deadline = CLOCK_FREQ / 40;
 444
 445             /* Additionnaly, we implicitly wait for the packetization time
 446              * multiplied by the number of missing packets. block is the first
 447              * non-missing packet (lowest sequence number). We have no better
 448              * estimated time of arrival, as we do not know the RTP timestamp
 449              * of not yet received packets. */
 450             deadline += block->i_pts;
 451             if (now >= deadline)
 452             {
 453                 rtp_decode (demux, session, src);
 454                 continue;
 455             }
 456             if (*deadlinep > deadline)
 457                 *deadlinep = deadline;
 458             pending = true; /* packet pending in buffer */
 459             break;
 460         }
 461     }
 462     return pending;
 463 }
 464
 465 /**
 466  * Dequeues all RTP packets and pass them to decoder. Not cancellation-safe(?).
 467  * This function can be used when the packet source is known not to reorder.
 468  */
 469 void rtp_dequeue_force (demux_t *demux, const rtp_session_t *session)
 470 {
 471     for (unsigned i = 0, max = session->srcc; i < max; i++)
 472     {
 473         rtp_source_t *src = session->srcv[i];
 474         block_t *block;
 475
 476         while (((block = src->blocks)) != NULL)
 477             rtp_decode (demux, session, src);
 478     }
 479 }
 480
 481 /**
 482  * Decodes one RTP packet.
 483  */
 484 static void
 485 rtp_decode (demux_t *demux, const rtp_session_t *session, rtp_source_t *src)
 486 {
 487     block_t *block = src->blocks;
 488
 489     assert (block);
 490     src->blocks = block->p_next;
 491     block->p_next = NULL;
 492
 493     /* Discontinuity detection */
 494     uint16_t delta_seq = rtp_seq (block) - (src->last_seq + 1);
 495     if (delta_seq != 0)
 496     {
 497         if (delta_seq >= 0x8000)
 498         {   /* Trash too late packets (and PIM Assert duplicates) */
 499             msg_Dbg (demux, "ignoring late packet (sequence: %"PRIu16")",
 500                       rtp_seq (block));
 501             goto drop;
 502         }
 503         msg_Warn (demux, "%"PRIu16" packet(s) lost", delta_seq);
 504         block->i_flags |= BLOCK_FLAG_DISCONTINUITY;
 505     }
 506     src->last_seq = rtp_seq (block);
 507
 508     /* Match the payload type */
 509     void *pt_data;
 510     const rtp_pt_t *pt = rtp_find_ptype (session, src, block, &pt_data);
 511     if (pt == NULL)
 512     {
 513         msg_Dbg (demux, "unknown payload (%"PRIu8")",
 514                  rtp_ptype (block));
 515         goto drop;
 516     }
 517
 518     if(pt->header)
 519         pt->header(demux, pt_data, block);
 520
 521     /* Computes the PTS from the RTP timestamp and payload RTP frequency.
 522      * DTS is unknown. Also, while the clock frequency depends on the payload
 523      * format, a single source MUST only use payloads of a chosen frequency.
 524      * Otherwise it would be impossible to compute consistent timestamps. */
 525     const uint32_t timestamp = rtp_timestamp (block);
 526     block->i_pts = src->ref_ntp
 527        + CLOCK_FREQ * (int32_t)(timestamp - src->ref_rtp) / pt->frequency;
 528     /* TODO: proper inter-medias/sessions sync (using RTCP-SR) */
 529     src->ref_ntp = block->i_pts;
 530     src->ref_rtp = timestamp;
 531
 532     /* CSRC count */
 533     size_t skip = 12u + (block->p_buffer[0] & 0x0F) * 4;
 534
 535     /* Extension header (ignored for now) */
 536     if (block->p_buffer[0] & 0x10)
 537     {
 538         skip += 4;
 539         if (block->i_buffer < skip)
 540             goto drop;
 541
 542         skip += 4 * GetWBE (block->p_buffer + skip - 2);
 543     }
 544
 545     if (block->i_buffer < skip)
 546         goto drop;
 547
 548     block->p_buffer += skip;
 549     block->i_buffer -= skip;
 550
 551     pt->decode (demux, pt_data, block);
 552     return;
 553
 554 drop:
 555     block_Release (block);
 556 }