aboutsummaryrefslogtreecommitdiffstats
path: root/src/jibuf.c
blob: 63769803fa58f2108deb8cdffc158e2638411b2c (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
/* (C) 2017 by Pau Espin Pedrol <pespin@sysmocom.de>
 *
 * 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 2 of the License, or
 * (at your option) any later version.
 */

#include <stdio.h>
#include <string.h>
#include <inttypes.h>

#include <osmocom/core/msgb.h>
#include <osmocom/core/timer.h>
#include <osmocom/core/talloc.h>
#include <osmocom/core/logging.h>
#include <osmocom/core/utils.h>

#include <osmocom/netif/amr.h>
#include <osmocom/netif/rtp.h>
#include <osmocom/netif/jibuf.h>

#include <arpa/inet.h>

/*! \addtogroup jibuf Osmocom Jitter Buffer
 *  @{
 */

/*! \file jibuf.c
 *  \brief Osmocom Jitter Buffer helpers
 */


/* Enable printing */
#define JIBUF_DEBUG 1
/* Sampling rate (in Hz) */
/* TODO: SAMPLE RATE can be guessed from rtp.p_type */
#define SAMPLE_RATE 8000


/* TUNABLE PARAMETERS: */

/* default {min,max}_delay values if set_{min,max}_delay() is never called */
#define JIBUF_DEFAULT_MIN_DELAY_MS 60
#define JIBUF_DEFAULT_MAX_DELAY_MS 200

/* How frequently (num of input packets) do we reselect a new reference? */
#define JIBUF_REFERENCE_TS_FREQ 60

/* How frequently (num of input packets) do we check if we should adapt the
 * buffer size (threshold_delay) ? */
#define JIBUF_BUFFER_RECALC_FREQ 40
/* How many pkts should be dropped at max every JIBUF_BUFFER_RECALC_FREQ input
 * pkts? */
#define JIBUF_ALLOWED_PKT_DROP 3
/* How many consecutive pkts can be dropped before triggering a buffer size incr ? */
#define JIBUF_ALLOWED_PKT_CONSECUTIVE_DROP 1
/* How much do we incr/decr the buffer size every time we recalculate it? */
#define JIBUF_BUFFER_INC_STEP 20
#define JIBUF_BUFFER_DEC_STEP 5


struct jibuf_msgb_cb {
	struct timeval ts;
	unsigned long *old_cb;
};
#define JIBUF_MSGB_CB(__msgb) ((struct jibuf_msgb_cb *)&((__msgb)->cb[0]))

void *tall_jibuf_ctx = NULL;


static void _ms2timeval(struct timeval *ts, uint32_t ms)
{
	ts->tv_sec = ms / 1000;
	ts->tv_usec = (ms % 1000) * 1000;
}

static uint32_t _timeval2ms(const struct timeval *ts)
{
	return ts->tv_sec * 1000 + ts->tv_usec / 1000;
}


static int clock_gettime_timeval(clockid_t clk_id, struct timeval *tp)
{
	struct timespec now;
	int n;

	n = clock_gettime(clk_id, &now);
	tp->tv_sec = now.tv_sec;
	tp->tv_usec = now.tv_nsec / 1000;

	return n;
}

static struct timeval *_msgb_scheduled_ts(const struct msgb *msg)
{
	struct jibuf_msgb_cb *jbcb = JIBUF_MSGB_CB(msg);
	return &jbcb->ts;
}

/* Add msgb to the list sorted by its scheduled output ts */
static void _llist_add_sorted(struct msgb *msg, struct llist_head *msg_list)
{
	struct msgb *cur;
	struct timeval *msg_ts = _msgb_scheduled_ts(msg);

	/* TODO: not sure if I need to use _safe here */
	llist_for_each_entry(cur, msg_list, list) {
		struct timeval *cur_ts = _msgb_scheduled_ts(cur);
		if (timercmp(msg_ts, cur_ts, <)) {
			__llist_add(&msg->list, cur->list.prev, &cur->list);
			return;
		}
	}

	/* we reached the end, add to the tail: */
	llist_add_tail(&msg->list, msg_list);

}


static bool _msg_get_marker(struct msgb *msg)
{
	/* TODO: make it more generic as a callback so that different types of
	 * pkts can be used ? */
	struct rtp_hdr *rtph = osmo_rtp_get_hdr(msg);
	return rtph->marker;
}

static uint16_t _msg_get_sequence(struct msgb *msg)
{
	struct rtp_hdr *rtph = osmo_rtp_get_hdr(msg);
	return ntohs(rtph->sequence);
}

static uint32_t _msg_get_timestamp(struct msgb *msg)
{
	struct rtp_hdr *rtph = osmo_rtp_get_hdr(msg);
	return ntohl(rtph->timestamp);
}

static int32_t _samples2ms(int32_t samples)
{
	/* XXX: SAMPLE RATE can be guessed from rtp.p_type */
	return samples * 1000 / SAMPLE_RATE;
}

/* Calculates pkt delay related to reference pkt. Similar concept to D(i,j) as
 * defined in RFC3550 (RTP). */
static int _calc_pkt_rel_delay(struct jibuf *jb, struct msgb *msg)
{
	uint32_t current_rx_ts = _timeval2ms(&jb->last_enqueue_time);
	uint32_t current_tx_ts = _msg_get_timestamp(msg);

	return _samples2ms((current_tx_ts - jb->ref_tx_ts)) - (current_rx_ts - jb->ref_rx_ts);
}


static void _msg_set_as_reference(struct jibuf *jb, struct msgb *msg)
{
	jb->ref_rx_ts = _timeval2ms(&jb->last_enqueue_time);
	jb->ref_tx_ts = _msg_get_timestamp(msg);
	jb->ref_tx_seq = _msg_get_sequence(msg);

# if JIBUF_DEBUG
	LOGP(DLMIB, LOGL_DEBUG, "New reference (seq=%"PRIu16" rx=%"PRIu32 \
		" tx=%"PRIu32")\n", jb->ref_tx_seq, jb->ref_rx_ts, jb->ref_tx_ts);
#endif
}

static void _dequeue_msg(struct jibuf *jb, struct msgb *msg)
{
	unsigned long *old_cb = JIBUF_MSGB_CB(msg)->old_cb;
	memcpy(msg->cb, old_cb, sizeof(msg->cb));
	talloc_free(old_cb);
	llist_del(&msg->list);

	jb->dequeue_cb(msg, jb->dequeue_cb_data);
}

static void _osmo_jibuf_timer_expired(void *data)
{
	struct jibuf *jb = (struct jibuf*) data;
		struct timeval delay_ts, now;
		struct msgb *msg, *next;

	llist_for_each_entry_safe(msg, next, &jb->msg_list, list) {
	struct timeval *msg_ts = _msgb_scheduled_ts(msg);
			clock_gettime_timeval(CLOCK_MONOTONIC, &now);
	if (timercmp(msg_ts, &now, >)) {
				jb->next_dequeue_time = *msg_ts;
				timersub(msg_ts, &now, &delay_ts);
				osmo_timer_schedule(&jb->timer,
					delay_ts.tv_sec, delay_ts.tv_usec);
				return;
			}

			_dequeue_msg(jb, msg);
	}

	/* XXX: maybe  try to tune the threshold based on the calculated output jitter? */
	/* XXX: try to find holes in the list and create fake pkts to improve the
		 jitter when packets do not arrive on time */
}


static void _recalc_threshold_delay(struct jibuf *jb)
{

	/* Recalculate every JIBUF_RECALC_FREQ_PKTS handled packets, or if we have too
		 many consecutive drops */
	uint32_t sum_pkts = jb->stats.total_enqueued + jb->stats.total_dropped +
								jb->last_dropped;

	if (jb->consecutive_drops <= JIBUF_ALLOWED_PKT_CONSECUTIVE_DROP &&
					sum_pkts % JIBUF_BUFFER_RECALC_FREQ != 0)
		return;

	if (jb->consecutive_drops > JIBUF_ALLOWED_PKT_CONSECUTIVE_DROP ||
					jb->last_dropped > JIBUF_ALLOWED_PKT_DROP)
		jb->threshold_delay = OSMO_MIN(
					jb->threshold_delay + JIBUF_BUFFER_INC_STEP,
					jb->max_delay);
	else
		jb->threshold_delay = OSMO_MAX(
					jb->threshold_delay - JIBUF_BUFFER_DEC_STEP,
					jb->min_delay);
# if JIBUF_DEBUG
	LOGP(DLMIB, LOGL_DEBUG, "New threshold: %u ms (freq=%d dropped=%d/%d consecutive=%d/%d)\n",
			jb->threshold_delay, JIBUF_BUFFER_RECALC_FREQ,
			jb->last_dropped, JIBUF_ALLOWED_PKT_DROP,
			jb->consecutive_drops, JIBUF_ALLOWED_PKT_CONSECUTIVE_DROP);
#endif

	jb->stats.total_dropped += jb->last_dropped;
	jb->last_dropped = 0;

}

//----------------------------------

/*! \brief Allocate a new jitter buffer instance
 *  \return the new allocated instance
 */
struct jibuf *osmo_jibuf_alloc()
{
	struct jibuf *jb;
	jb = talloc_zero(tall_jibuf_ctx, struct jibuf);

	jb->min_delay = JIBUF_DEFAULT_MIN_DELAY_MS;
	jb->max_delay = JIBUF_DEFAULT_MAX_DELAY_MS;
	jb->threshold_delay = jb->min_delay;

	INIT_LLIST_HEAD(&jb->msg_list);

	jb->timer.cb = _osmo_jibuf_timer_expired;
	jb->timer.data = jb;

	return jb;
}

/*! \brief Destroy a previously allocated jitter buffer instance
 *  \param[in] jb Previously allocated (non-null) jitter buffer instance
 *
 * All the queued packets are dequeued before deleting the instance.
 */
void osmo_jibuf_delete(struct jibuf *jb)
{
	struct msgb *msg, *tmp;
	osmo_timer_del(&jb->timer);
	llist_for_each_entry_safe(msg, tmp, &jb->msg_list, list)
	_dequeue_msg(jb, msg);

	talloc_free(jb);
}

/*! \brief Try to enqueue a packet into the jitter buffer
 *  \param[in] jb jitter buffer instance
 *  \param[in] msg msgb to enqueue, containing an RTP packet
 *  \return <0 if the packet was dropped, 0 otherwise
 *
 * This function calculates the delay for the enqueued packet. If the delay is
 * bigger than the current buffer size, the function returns -1 and the caller
 * owns the packet again and can free it if required. If the packet is enqueued,
 * 0 is returned and the exact same packet (ownership transfer, no copy is made)
 * will be available again through the dequeue_cb() when the queue timer for
 * this packet expires.
 */
int osmo_jibuf_enqueue(struct jibuf *jb, struct msgb *msg)
{
	int rel_delay, delay;
	struct timeval delay_ts, sched_ts;

	clock_gettime_timeval(CLOCK_MONOTONIC, &jb->last_enqueue_time);

	/* If packet contains a mark -> start of talkspurt.
	 * A lot of packets may have been suppressed by the sender before it,
	 * so let's take it as a reference
	 */
	if (!jb->started || _msg_get_marker(msg)) {
		jb->started = true;
		_msg_set_as_reference(jb, msg);
		rel_delay = 0;
	} else {
		rel_delay = _calc_pkt_rel_delay(jb, msg);
	}

	/* Avoid time skew with sender (or drop-everything state),
	   reselect a new reference from time to time */
	//if ((int)(_msg_get_sequence(msg) - jb->ref_tx_seq) > JIBUF_REFERENCE_TS_FREQ)
	//	_msg_set_as_reference(jb, msg);

	delay = jb->threshold_delay + rel_delay;

	/* packet too late, let's drop it and incr buffer size if encouraged */
	if (delay < 0) {
		jb->last_dropped++;
		jb->consecutive_drops++;
#if JIBUF_DEBUG
		LOGP(DLMIB, LOGL_DEBUG, "dropped %u > %u (seq=%"PRIu16" ts=%"PRIu32")\n",
			rel_delay, jb->threshold_delay, _msg_get_sequence(msg),
			_msg_get_timestamp(msg));
#endif
		_recalc_threshold_delay(jb);
		return -1;
	} else {
		jb->consecutive_drops = 0;
		jb->stats.total_enqueued++;
	}

	_ms2timeval(&delay_ts, (uint32_t) delay);
	timeradd(&jb->last_enqueue_time, &delay_ts, &sched_ts);

#if JIBUF_DEBUG
	LOGP(DLMIB, LOGL_DEBUG, "enqueuing packet seq=%"PRIu16" rel=%d delay=%d" \
		" thres=%d {%lu.%06lu -> %lu.%06lu} %s\n",
		_msg_get_sequence(msg), rel_delay, delay, jb->threshold_delay,
		jb->last_enqueue_time.tv_sec, jb->last_enqueue_time.tv_usec,
		sched_ts.tv_sec, sched_ts.tv_usec, _msg_get_marker(msg)? "M" : "");
#endif

	/* Add scheduled dequeue time in msg->cb so we can check it later */
	unsigned long *old_cb = talloc_memdup(tall_jibuf_ctx, msg->cb, sizeof(msg->cb));
	struct jibuf_msgb_cb *jbcb = JIBUF_MSGB_CB(msg);
	jbcb->ts = sched_ts;
	jbcb->old_cb = old_cb;

	_llist_add_sorted(msg, &jb->msg_list);


	/* See if updating the timer is needed: */
	if (!osmo_timer_pending(&jb->timer) ||
			timercmp(&sched_ts, &jb->next_dequeue_time, <))  {
		jb->next_dequeue_time = sched_ts;
		osmo_timer_schedule(&jb->timer, 0, delay * 1000);
	}

	/* Let's check packet loss stats to see if buffer_size must be changed */
	_recalc_threshold_delay(jb);

	return 0;
}

/*! \brief Check whether the jitter buffer instance has packets queued or not.
 *  \param[in] jb jitter buffer instance
 *  \return true if the queue is empty, false otherwise.
 */
bool osmo_jibuf_empty(struct jibuf *jb)
{
	return llist_empty(&jb->msg_list);
}


/*! \brief Set minimum buffer size for the jitter buffer
 *  \param[in] jb jitter buffer instance
 *  \param[in] min_delay Minimum buffer size, as in minimum delay in milliseconds
 */
void osmo_jibuf_set_min_delay(struct jibuf *jb, uint32_t min_delay)
{
	jb->min_delay = min_delay ? min_delay : JIBUF_DEFAULT_MIN_DELAY_MS;
	jb->threshold_delay = OSMO_MAX(jb->min_delay, jb->threshold_delay);
}

/*! \brief Set maximum buffer size for the jitter buffer
 *  \param[in] jb jitter buffer instance
 *  \param[in] max_delay Maximum buffer size, as in maximum delay in milliseconds
 */
void osmo_jibuf_set_max_delay(struct jibuf *jb, uint32_t max_delay)
{
	jb->max_delay = max_delay ? max_delay : JIBUF_DEFAULT_MAX_DELAY_MS;
	jb->threshold_delay = OSMO_MIN(jb->max_delay, jb->threshold_delay);
}

/*! \brief Set dequeue callback for the jitter buffer
 *  \param[in] jb jitter buffer instance
 *  \param[in] dequeue_cb function pointer to call back when the dequeue timer for a given packet expires
 *  \param[in] cb_data data pointer to be passed to dequeue_cb together with the msgb.
 */
void osmo_jibuf_set_dequeue_cb(struct jibuf *jb, osmo_jibuf_dequeue_cb
						dequeue_cb, void* cb_data)
{
	jb->dequeue_cb = dequeue_cb;
	jb->dequeue_cb_data = cb_data;
}


/*! @} */