RTP: try to improve packet re-ordering
[vlc.git] / 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 General Public License
10  * as published by the Free Software Foundation; either version 2.0
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 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/vlc.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->frequency = pt->frequency;
127     ppt->number = pt->number;
128     msg_Dbg (demux, "added payload type %"PRIu8" (f = %"PRIu32" Hz)",
129              ppt->number, ppt->frequency);
130
131     assert (ppt->frequency > 0); /* SIGFPE! */
132     (void)demux;
133     return 0;
134 }
135
136 /** State for an RTP source */
137 struct rtp_source_t
138 {
139     uint32_t ssrc;
140     uint32_t jitter;  /* interarrival delay jitter estimate */
141     mtime_t  last_rx; /* last received packet local timestamp */
142     uint32_t last_ts; /* last received packet RTP timestamp */
143
144     uint16_t bad_seq; /* tentatively next expected sequence for resync */
145     uint16_t max_seq; /* next expected sequence */
146
147     uint16_t last_seq; /* sequence of the last dequeued packet */
148     block_t *blocks; /* re-ordered blocks queue */
149     void    *opaque[0]; /* Per-source private payload data */
150 };
151
152 /**
153  * Initializes a new RTP source within an RTP session.
154  */
155 static rtp_source_t *
156 rtp_source_create (demux_t *demux, const rtp_session_t *session,
157                    uint32_t ssrc, uint16_t init_seq)
158 {
159     rtp_source_t *source;
160
161     source = malloc (sizeof (*source) + (sizeof (void *) * session->ptc));
162     if (source == NULL)
163         return NULL;
164
165     source->ssrc = ssrc;
166     source->jitter = 0;
167     source->max_seq = source->bad_seq = init_seq;
168     source->last_seq = init_seq - 1;
169     source->blocks = NULL;
170
171     /* Initializes all payload */
172     for (unsigned i = 0; i < session->ptc; i++)
173         source->opaque[i] = session->ptv[i].init (demux);
174
175     msg_Dbg (demux, "added RTP source (%08x)", ssrc);
176     return source;
177 }
178
179
180 /**
181  * Destroys an RTP source and its associated streams.
182  */
183 static void
184 rtp_source_destroy (demux_t *demux, const rtp_session_t *session,
185                     rtp_source_t *source)
186 {
187     msg_Dbg (demux, "removing RTP source (%08x)", source->ssrc);
188
189     for (unsigned i = 0; i < session->ptc; i++)
190         session->ptv[i].destroy (demux, source->opaque[i]);
191     block_ChainRelease (source->blocks);
192     free (source);
193 }
194
195 static inline uint16_t rtp_seq (const block_t *block)
196 {
197     assert (block->i_buffer >= 4);
198     return GetWBE (block->p_buffer + 2);
199 }
200
201 static inline uint32_t rtp_timestamp (const block_t *block)
202 {
203     assert (block->i_buffer >= 12);
204     return GetDWBE (block->p_buffer + 4);
205 }
206
207 static const struct rtp_pt_t *
208 rtp_find_ptype (const rtp_session_t *session, rtp_source_t *source,
209                 const block_t *block, void **pt_data)
210 {
211     uint8_t ptype = rtp_ptype (block);
212
213     for (unsigned i = 0; i < session->ptc; i++)
214     {
215         if (session->ptv[i].number == ptype)
216         {
217             if (pt_data != NULL)
218                 *pt_data = source->opaque[i];
219             return &session->ptv[i];
220         }
221     }
222     return NULL;
223 }
224
225 /**
226  * Receives an RTP packet and queues it.
227  * @param demux VLC demux object
228  * @param session RTP session receiving the packet
229  * @param block RTP packet including the RTP header
230  */
231 void
232 rtp_queue (demux_t *demux, rtp_session_t *session, block_t *block)
233 {
234     demux_sys_t *p_sys = demux->p_sys;
235
236     /* RTP header sanity checks (see RFC 3550) */
237     if (block->i_buffer < 12)
238         goto drop;
239     if ((block->p_buffer[0] >> 6 ) != 2) /* RTP version number */
240         goto drop;
241
242     /* Remove padding if present */
243     if (block->p_buffer[0] & 0x20)
244     {
245         uint8_t padding = block->p_buffer[block->i_buffer - 1];
246         if ((padding == 0) || (block->i_buffer < (12u + padding)))
247             goto drop; /* illegal value */
248
249         block->i_buffer -= padding;
250     }
251
252     mtime_t        now = mdate ();
253     rtp_source_t  *src  = NULL;
254     const uint16_t seq  = GetWBE (block->p_buffer + 2);
255     const uint32_t ssrc = GetDWBE (block->p_buffer + 8);
256
257     /* In most case, we know this source already */
258     for (unsigned i = 0, max = session->srcc; i < max; i++)
259     {
260         rtp_source_t *tmp = session->srcv[i];
261         if (tmp->ssrc == ssrc)
262         {
263             src = tmp;
264             break;
265         }
266
267         /* RTP source garbage collection */
268         if ((tmp->last_rx + (p_sys->timeout * CLOCK_FREQ)) < now)
269         {
270             rtp_source_destroy (demux, session, tmp);
271             if (--session->srcc > 0)
272                 session->srcv[i] = session->srcv[session->srcc - 1];
273         }
274     }
275
276     if (src == NULL)
277     {
278         /* New source */
279         if (session->srcc >= p_sys->max_src)
280         {
281             msg_Warn (demux, "too many RTP sessions");
282             goto drop;
283         }
284
285         rtp_source_t **tab;
286         tab = realloc (session->srcv, (session->srcc + 1) * sizeof (*tab));
287         if (tab == NULL)
288             goto drop;
289         session->srcv = tab;
290
291         src = rtp_source_create (demux, session, ssrc, seq);
292         if (src == NULL)
293             goto drop;
294
295         tab[session->srcc++] = src;
296         /* Cannot compute jitter yet */
297     }
298     else
299     {
300         const rtp_pt_t *pt = rtp_find_ptype (session, src, block, NULL);
301
302         if (pt != NULL)
303         {
304             /* Recompute jitter estimate.
305              * That is computed from the RTP timestamps and the system clock.
306              * It is independent of RTP sequence. */
307             uint32_t freq = pt->frequency;
308             uint32_t ts = rtp_timestamp (block);
309             int64_t d = ((now - src->last_rx) * freq) / CLOCK_FREQ;
310             d        -=    ts - src->last_ts;
311             if (d < 0) d = -d;
312             src->jitter += ((d - src->jitter) + 8) >> 4;
313         }
314     }
315     src->last_rx = now;
316     src->last_ts = rtp_timestamp (block);
317
318     /* Be optimistic for the first packet. Certain codec, such as Vorbis
319      * do not like loosing the first packet(s), so we cannot just wait
320      * for proper sequence synchronization. And we don't want to assume that
321      * the sender starts at seq=0 either. */
322     if (src->blocks == NULL)
323         src->max_seq = seq - p_sys->max_dropout;
324
325     /* Check sequence number */
326     /* NOTE: the sequence number is per-source,
327      * but is independent from the payload type. */
328     uint16_t delta_seq = seq - (src->max_seq + 1);
329     if ((delta_seq < 0x8000) ? (delta_seq > p_sys->max_dropout)
330                              : ((65535 - delta_seq) > p_sys->max_misorder))
331     {
332         msg_Dbg (demux, "sequence discontinuity (got: %u, expected: %u)",
333                  seq, (src->max_seq + 1) & 0xffff);
334         if (seq == ((src->bad_seq + 1) & 0xffff))
335         {
336             src->max_seq = src->bad_seq = seq;
337             msg_Warn (demux, "sequence resynchronized");
338             block_ChainRelease (src->blocks);
339             src->blocks = NULL;
340         }
341         else
342         {
343             src->bad_seq = seq;
344             goto drop;
345         }
346     }
347     else
348     if (delta_seq < 0x8000)
349         src->max_seq = seq;
350
351     /* Queues the block in sequence order,
352      * hence there is a single queue for all payload types. */
353     block_t **pp = &src->blocks;
354     for (block_t *prev = *pp; prev != NULL; prev = *pp)
355     {
356         int16_t delta_seq = seq - rtp_seq (prev);
357         if (delta_seq < 0)
358             break;
359         if (delta_seq == 0)
360             goto drop; /* duplicate */
361         pp = &prev->p_next;
362     }
363     block->p_next = *pp;
364     *pp = block;
365
366     /*rtp_decode (demux, session, src);*/
367     return;
368
369 drop:
370     block_Release (block);
371 }
372
373
374 static void
375 rtp_decode (demux_t *demux, const rtp_session_t *session, rtp_source_t *src)
376 {
377     block_t *block = src->blocks;
378
379     assert (block);
380     src->blocks = block->p_next;
381     block->p_next = NULL;
382
383     /* Discontinuity detection */
384     uint16_t delta_seq = rtp_seq (block) - (src->last_seq + 1);
385     if (delta_seq != 0)
386     {
387         if (delta_seq >= 0x8000)
388         {   /* Unrecoverable if later packets have already been dequeued */
389             msg_Warn (demux, "ignoring late packet (sequence: %u)",
390                       rtp_seq (block));
391             goto drop;
392         }
393         block->i_flags |= BLOCK_FLAG_DISCONTINUITY;
394     }
395     src->last_seq = rtp_seq (block);
396
397     /* Match the payload type */
398     void *pt_data;
399     const rtp_pt_t *pt = rtp_find_ptype (session, src, block, &pt_data);
400     if (pt == NULL)
401     {
402         msg_Dbg (demux, "ignoring unknown payload (%"PRIu8")",
403                  rtp_ptype (block));
404         goto drop;
405     }
406
407     /* Computes the PTS from the RTP timestamp and payload RTP frequency.
408      * DTS is unknown. Also, while the clock frequency depends on the payload
409      * format, a single source MUST only use payloads of a chosen frequency.
410      * Otherwise it would be impossible to compute consistent timestamps. */
411     /* FIXME: handle timestamp wrap properly */
412     /* TODO: inter-medias/sessions sync (using RTCP-SR) */
413     const uint32_t timestamp = rtp_timestamp (block);
414     block->i_pts = UINT64_C(1) * CLOCK_FREQ * timestamp / pt->frequency;
415
416     /* CSRC count */
417     size_t skip = 12u + (block->p_buffer[0] & 0x0F) * 4;
418
419     /* Extension header (ignored for now) */
420     if (block->p_buffer[0] & 0x10)
421     {
422         skip += 4;
423         if (block->i_buffer < skip)
424             goto drop;
425
426         skip += 4 * GetWBE (block->p_buffer + skip - 2);
427     }
428
429     if (block->i_buffer < skip)
430         goto drop;
431
432     block->p_buffer += skip;
433     block->i_buffer -= skip;
434
435     pt->decode (demux, pt_data, block);
436     return;
437
438 drop:
439     block_Release (block);
440 }
441
442
443 bool rtp_dequeue (demux_t *demux, const rtp_session_t *session,
444                   mtime_t *restrict deadlinep)
445 {
446     mtime_t now = mdate ();
447     bool pending = false;
448
449     for (unsigned i = 0, max = session->srcc; i < max; i++)
450     {
451         rtp_source_t *src = session->srcv[i];
452         block_t *block;
453
454         /* Because of IP packet delay variation (IPDV), we need to guesstimate
455          * how long to wait for a missing packet in the RTP sequence
456          * (see RFC3393 for background on IPDV).
457          *
458          * This situation occurs if a packet got lost, or if the network has
459          * re-ordered packets. Unfortunately, the MSL is 2 minutes, orders of
460          * magnitude too long for multimedia. We need a tradeoff.
461          * If we underestimated IPDV, we may have to discard valid but late
462          * packets. If we overestimate it, we will either cause too much
463          * delay, or worse, underflow our downstream buffers, as we wait for
464          * definitely a lost packets.
465          *
466          * The rest of the "de-jitter buffer" work is done by the interval
467          * LibVLC E/S-out clock synchronization. Here, we need to bother about
468          * re-ordering packets, as decoders can't cope with mis-ordered data.
469          */
470         while (((block = src->blocks)) != NULL)
471         {
472 #if 0
473             if (rtp_seq (block) == ((src->last_seq + 1) & 0xffff))
474             {   /* Next block ready, no need to wait */
475                 rtp_decode (demux, session, src);
476                 continue;
477             }
478 #endif
479             /* Wait for 3 times the inter-arrival delay variance (about 99.7%
480              * match for random gaussian jitter). Additionnaly, we implicitly
481              * wait for misordering times the packetization time.
482              */
483             mtime_t deadline = src->last_rx;
484             const rtp_pt_t *pt = rtp_find_ptype (session, src, block, NULL);
485             if (pt)
486                 deadline += UINT64_C(3) * CLOCK_FREQ * src->jitter
487                             / pt->frequency;
488
489             if (now >= deadline)
490             {
491                 rtp_decode (demux, session, src);
492                 continue;
493             }
494             if (*deadlinep > deadline)
495                 *deadlinep = deadline;
496             pending = true; /* packet pending in buffer */
497             break;
498         }
499     }
500     return pending;
501 }