1/***
2  This file is part of PulseAudio.
3
4  Copyright 2006 Lennart Poettering
5
6  PulseAudio is free software; you can redistribute it and/or modify
7  it under the terms of the GNU Lesser General Public License as published
8  by the Free Software Foundation; either version 2.1 of the License,
9  or (at your option) any later version.
10
11  PulseAudio is distributed in the hope that it will be useful, but
12  WITHOUT ANY WARRANTY; without even the implied warranty of
13  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
14  General Public License for more details.
15
16  You should have received a copy of the GNU Lesser General Public License
17  along with PulseAudio; if not, see <http://www.gnu.org/licenses/>.
18***/
19
20#ifdef HAVE_CONFIG_H
21#include <config.h>
22#endif
23
24#include <stdlib.h>
25#include <string.h>
26#include <errno.h>
27#include <unistd.h>
28#include <sys/ioctl.h>
29
30#ifdef HAVE_SYS_FILIO_H
31#include <sys/filio.h>
32#endif
33
34#ifdef HAVE_SYS_UIO_H
35#include <sys/uio.h>
36#endif
37
38#include <pulsecore/core-error.h>
39#include <pulsecore/log.h>
40#include <pulsecore/macro.h>
41#include <pulsecore/core-util.h>
42#include <pulsecore/arpa-inet.h>
43#include <pulsecore/poll.h>
44
45#include "rtp.h"
46
47typedef struct pa_rtp_context {
48    int fd;
49    uint16_t sequence;
50    uint32_t timestamp;
51    uint32_t ssrc;
52    uint8_t payload;
53    size_t frame_size;
54    size_t mtu;
55
56    uint8_t *recv_buf;
57    size_t recv_buf_size;
58    pa_memchunk memchunk;
59} pa_rtp_context;
60
61pa_rtp_context* pa_rtp_context_new_send(int fd, uint8_t payload, size_t mtu, const pa_sample_spec *ss, bool enable_opus) {
62    pa_rtp_context *c;
63
64    pa_assert(fd >= 0);
65
66    pa_log_info("Initialising native RTP backend for send");
67
68    c = pa_xnew0(pa_rtp_context, 1);
69
70    c->fd = fd;
71    c->sequence = (uint16_t) (rand()*rand());
72    c->timestamp = 0;
73    c->ssrc = (uint32_t) (rand()*rand());
74    c->payload = (uint8_t) (payload & 127U);
75    c->frame_size = pa_frame_size(ss);
76    c->mtu = mtu;
77
78    c->recv_buf = NULL;
79    c->recv_buf_size = 0;
80    pa_memchunk_reset(&c->memchunk);
81
82    return c;
83}
84
85#define MAX_IOVECS 16
86
87int pa_rtp_send(pa_rtp_context *c, pa_memblockq *q) {
88    struct iovec iov[MAX_IOVECS];
89    pa_memblock* mb[MAX_IOVECS];
90    int iov_idx = 1;
91    size_t n = 0;
92
93    pa_assert(c);
94    pa_assert(q);
95
96    if (pa_memblockq_get_length(q) < c->mtu)
97        return 0;
98
99    for (;;) {
100        int r;
101        pa_memchunk chunk;
102
103        pa_memchunk_reset(&chunk);
104
105        if ((r = pa_memblockq_peek(q, &chunk)) >= 0) {
106
107            size_t k = n + chunk.length > c->mtu ? c->mtu - n : chunk.length;
108
109            pa_assert(chunk.memblock);
110
111            iov[iov_idx].iov_base = pa_memblock_acquire_chunk(&chunk);
112            iov[iov_idx].iov_len = k;
113            mb[iov_idx] = chunk.memblock;
114            iov_idx ++;
115
116            n += k;
117            pa_memblockq_drop(q, k);
118        }
119
120        pa_assert(n % c->frame_size == 0);
121
122        if (r < 0 || n >= c->mtu || iov_idx >= MAX_IOVECS) {
123            uint32_t header[3];
124            struct msghdr m;
125            ssize_t k;
126            int i;
127
128            if (n > 0) {
129                header[0] = htonl(((uint32_t) 2 << 30) | ((uint32_t) c->payload << 16) | ((uint32_t) c->sequence));
130                header[1] = htonl(c->timestamp);
131                header[2] = htonl(c->ssrc);
132
133                iov[0].iov_base = (void*)header;
134                iov[0].iov_len = sizeof(header);
135
136                m.msg_name = NULL;
137                m.msg_namelen = 0;
138                m.msg_iov = iov;
139                m.msg_iovlen = (size_t) iov_idx;
140                m.msg_control = NULL;
141                m.msg_controllen = 0;
142                m.msg_flags = 0;
143
144                k = sendmsg(c->fd, &m, MSG_DONTWAIT);
145
146                for (i = 1; i < iov_idx; i++) {
147                    pa_memblock_release(mb[i]);
148                    pa_memblock_unref(mb[i]);
149                }
150
151                c->sequence++;
152            } else
153                k = 0;
154
155            c->timestamp += (unsigned) (n/c->frame_size);
156
157            if (k < 0) {
158                if (errno != EAGAIN && errno != EINTR) /* If the queue is full, just ignore it */
159                    pa_log("sendmsg() failed: %s", pa_cstrerror(errno));
160                return -1;
161            }
162
163            if (r < 0 || pa_memblockq_get_length(q) < c->mtu)
164                break;
165
166            n = 0;
167            iov_idx = 1;
168        }
169    }
170
171    return 0;
172}
173
174pa_rtp_context* pa_rtp_context_new_recv(int fd, uint8_t payload, const pa_sample_spec *ss, bool enable_opus) {
175    pa_rtp_context *c;
176
177    pa_log_info("Initialising native RTP backend for receive");
178
179    c = pa_xnew0(pa_rtp_context, 1);
180
181    c->fd = fd;
182    c->payload = payload;
183    c->frame_size = pa_frame_size(ss);
184
185    c->recv_buf_size = 2000;
186    c->recv_buf = pa_xmalloc(c->recv_buf_size);
187    pa_memchunk_reset(&c->memchunk);
188
189    return c;
190}
191
192int pa_rtp_recv(pa_rtp_context *c, pa_memchunk *chunk, pa_mempool *pool, uint32_t *rtp_tstamp, struct timeval *tstamp) {
193    int size;
194    size_t audio_length;
195    size_t metadata_length;
196    struct msghdr m;
197    struct cmsghdr *cm;
198    struct iovec iov;
199    uint32_t header;
200    uint32_t ssrc;
201    uint8_t payload;
202    unsigned cc;
203    ssize_t r;
204    uint8_t aux[1024];
205    bool found_tstamp = false;
206
207    pa_assert(c);
208    pa_assert(chunk);
209
210    pa_memchunk_reset(chunk);
211
212    if (ioctl(c->fd, FIONREAD, &size) < 0) {
213        pa_log_warn("FIONREAD failed: %s", pa_cstrerror(errno));
214        goto fail;
215    }
216
217    if (size <= 0) {
218        /* size can be 0 due to any of the following reasons:
219         *
220         * 1. Somebody sent us a perfectly valid zero-length UDP packet.
221         * 2. Somebody sent us a UDP packet with a bad CRC.
222         *
223         * It is unknown whether size can actually be less than zero.
224         *
225         * In the first case, the packet has to be read out, otherwise the
226         * kernel will tell us again and again about it, thus preventing
227         * reception of any further packets. So let's just read it out
228         * now and discard it later, when comparing the number of bytes
229         * received (0) with the number of bytes wanted (1, see below).
230         *
231         * In the second case, recvmsg() will fail, thus allowing us to
232         * return the error.
233         *
234         * Just to avoid passing zero-sized memchunks and NULL pointers to
235         * recvmsg(), let's force allocation of at least one byte by setting
236         * size to 1.
237         */
238        size = 1;
239    }
240
241    if (c->recv_buf_size < (size_t) size) {
242        do
243            c->recv_buf_size *= 2;
244        while (c->recv_buf_size < (size_t) size);
245
246        c->recv_buf = pa_xrealloc(c->recv_buf, c->recv_buf_size);
247    }
248
249    pa_assert(c->recv_buf_size >= (size_t) size);
250
251    iov.iov_base = c->recv_buf;
252    iov.iov_len = (size_t) size;
253
254    m.msg_name = NULL;
255    m.msg_namelen = 0;
256    m.msg_iov = &iov;
257    m.msg_iovlen = 1;
258    m.msg_control = aux;
259    m.msg_controllen = sizeof(aux);
260    m.msg_flags = 0;
261
262    r = recvmsg(c->fd, &m, 0);
263
264    if (r != size) {
265        if (r < 0 && errno != EAGAIN && errno != EINTR)
266            pa_log_warn("recvmsg() failed: %s", r < 0 ? pa_cstrerror(errno) : "size mismatch");
267
268        goto fail;
269    }
270
271    if (size < 12) {
272        pa_log_warn("RTP packet too short.");
273        goto fail;
274    }
275
276    memcpy(&header, iov.iov_base, sizeof(uint32_t));
277    memcpy(rtp_tstamp, (uint8_t*) iov.iov_base + 4, sizeof(uint32_t));
278    memcpy(&ssrc, (uint8_t*) iov.iov_base + 8, sizeof(uint32_t));
279
280    header = ntohl(header);
281    *rtp_tstamp = ntohl(*rtp_tstamp);
282    ssrc = ntohl(c->ssrc);
283
284    if ((header >> 30) != 2) {
285        pa_log_warn("Unsupported RTP version.");
286        goto fail;
287    }
288
289    if ((header >> 29) & 1) {
290        pa_log_warn("RTP padding not supported.");
291        goto fail;
292    }
293
294    if ((header >> 28) & 1) {
295        pa_log_warn("RTP header extensions not supported.");
296        goto fail;
297    }
298
299    if (ssrc != c->ssrc) {
300        pa_log_debug("Got unexpected SSRC");
301        goto fail;
302    }
303
304    cc = (header >> 24) & 0xF;
305    payload = (uint8_t) ((header >> 16) & 127U);
306    c->sequence = (uint16_t) (header & 0xFFFFU);
307
308    metadata_length = 12 + cc * 4;
309
310    if (payload != c->payload) {
311        pa_log_debug("Got unexpected payload: %u", payload);
312        goto fail;
313    }
314
315    if (metadata_length > (unsigned) size) {
316        pa_log_warn("RTP packet too short. (CSRC)");
317        goto fail;
318    }
319
320    audio_length = size - metadata_length;
321
322    if (audio_length % c->frame_size != 0) {
323        pa_log_warn("Bad RTP packet size.");
324        goto fail;
325    }
326
327    if (c->memchunk.length < (unsigned) audio_length) {
328        size_t l;
329
330        if (c->memchunk.memblock)
331            pa_memblock_unref(c->memchunk.memblock);
332
333        l = PA_MAX((size_t) audio_length, pa_mempool_block_size_max(pool));
334
335        c->memchunk.memblock = pa_memblock_new(pool, l);
336        c->memchunk.index = 0;
337        c->memchunk.length = pa_memblock_get_length(c->memchunk.memblock);
338    }
339
340    memcpy(pa_memblock_acquire_chunk(&c->memchunk), c->recv_buf + metadata_length, audio_length);
341    pa_memblock_release(c->memchunk.memblock);
342
343    chunk->memblock = pa_memblock_ref(c->memchunk.memblock);
344    chunk->index = c->memchunk.index;
345    chunk->length = audio_length;
346
347    c->memchunk.index += audio_length;
348    c->memchunk.length -= audio_length;
349
350    if (c->memchunk.length <= 0) {
351        pa_memblock_unref(c->memchunk.memblock);
352        pa_memchunk_reset(&c->memchunk);
353    }
354
355    for (cm = CMSG_FIRSTHDR(&m); cm; cm = CMSG_NXTHDR(&m, cm))
356        if (cm->cmsg_level == SOL_SOCKET && cm->cmsg_type == SCM_TIMESTAMP) {
357            memcpy(tstamp, CMSG_DATA(cm), sizeof(struct timeval));
358            found_tstamp = true;
359            break;
360        }
361
362    if (!found_tstamp) {
363        pa_log_warn("Couldn't find SCM_TIMESTAMP data in auxiliary recvmsg() data!");
364        pa_zero(*tstamp);
365    }
366
367    return 0;
368
369fail:
370    if (chunk->memblock)
371        pa_memblock_unref(chunk->memblock);
372
373    return -1;
374}
375
376void pa_rtp_context_free(pa_rtp_context *c) {
377    pa_assert(c);
378
379    pa_assert_se(pa_close(c->fd) == 0);
380
381    if (c->memchunk.memblock)
382        pa_memblock_unref(c->memchunk.memblock);
383
384    pa_xfree(c->recv_buf);
385    pa_xfree(c);
386}
387
388size_t pa_rtp_context_get_frame_size(pa_rtp_context *c) {
389    return c->frame_size;
390}
391
392pa_rtpoll_item* pa_rtp_context_get_rtpoll_item(pa_rtp_context *c, pa_rtpoll *rtpoll) {
393    pa_rtpoll_item *item;
394    struct pollfd *p;
395
396    item = pa_rtpoll_item_new(rtpoll, PA_RTPOLL_LATE, 1);
397
398    p = pa_rtpoll_item_get_pollfd(item, NULL);
399    p->fd = c->fd;
400    p->events = POLLIN;
401    p->revents = 0;
402
403    return item;
404}
405