?  search:
summary | log | graphiclog1 | graphiclog2 | commit | commitdiff | tree | refs | edit | fork
blame | history | raw | HEAD
build-sys: bump soname
[pulseaudio-mirror.git] / src / pulsecore / pstream.c
blob3e0bfa3beb58da9b6277f0d1867a93615296b418
1 /***
2 This file is part of PulseAudio.
3
4 Copyright 2004-2006 Lennart Poettering
5 Copyright 2006 Pierre Ossman <ossman@cendio.se> for Cendio AB
6
7 PulseAudio is free software; you can redistribute it and/or modify
8 it under the terms of the GNU Lesser General Public License as
9 published by the Free Software Foundation; either version 2.1 of the
10 License, or (at your option) any later version.
11
12 PulseAudio is distributed in the hope that it will be useful, but
13 WITHOUT ANY WARRANTY; without even the implied warranty of
14 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
15 Lesser General Public License for more details.
16
17 You should have received a copy of the GNU Lesser General Public
18 License along with PulseAudio; if not, write to the Free Software
19 Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307
20 USA.
21 ***/
22
23 #ifdef HAVE_CONFIG_H
24 #include <config.h>
25 #endif
26
27 #include <stdio.h>
28 #include <stdlib.h>
29 #include <unistd.h>
30
31 #ifdef HAVE_SYS_SOCKET_H
32 #include <sys/socket.h>
33 #endif
34 #ifdef HAVE_SYS_UN_H
35 #include <sys/un.h>
36 #endif
37 #ifdef HAVE_NETINET_IN_H
38 #include <netinet/in.h>
39 #endif
40
41
42 #include <pulse/xmalloc.h>
43
44 #include <pulsecore/winsock.h>
45 #include <pulsecore/queue.h>
46 #include <pulsecore/log.h>
47 #include <pulsecore/core-scache.h>
48 #include <pulsecore/creds.h>
49 #include <pulsecore/refcnt.h>
50 #include <pulsecore/flist.h>
51 #include <pulsecore/macro.h>
52
53 #include "pstream.h"
54
55 /* We piggyback information if audio data blocks are stored in SHM on the seek mode */
56 #define PA_FLAG_SHMDATA 0x80000000LU
57 #define PA_FLAG_SHMRELEASE 0x40000000LU
58 #define PA_FLAG_SHMREVOKE 0xC0000000LU
59 #define PA_FLAG_SHMMASK 0xFF000000LU
60 #define PA_FLAG_SEEKMASK 0x000000FFLU
61
62 /* The sequence descriptor header consists of 5 32bit integers: */
63 enum {
64 PA_PSTREAM_DESCRIPTOR_LENGTH,
65 PA_PSTREAM_DESCRIPTOR_CHANNEL,
66 PA_PSTREAM_DESCRIPTOR_OFFSET_HI,
67 PA_PSTREAM_DESCRIPTOR_OFFSET_LO,
68 PA_PSTREAM_DESCRIPTOR_FLAGS,
69 PA_PSTREAM_DESCRIPTOR_MAX
70 };
71
72 /* If we have an SHM block, this info follows the descriptor */
73 enum {
74 PA_PSTREAM_SHM_BLOCKID,
75 PA_PSTREAM_SHM_SHMID,
76 PA_PSTREAM_SHM_INDEX,
77 PA_PSTREAM_SHM_LENGTH,
78 PA_PSTREAM_SHM_MAX
79 };
80
81 typedef uint32_t pa_pstream_descriptor[PA_PSTREAM_DESCRIPTOR_MAX];
82
83 #define PA_PSTREAM_DESCRIPTOR_SIZE (PA_PSTREAM_DESCRIPTOR_MAX*sizeof(uint32_t))
84 #define FRAME_SIZE_MAX_ALLOW PA_SCACHE_ENTRY_SIZE_MAX /* allow uploading a single sample in one frame at max */
85
86 PA_STATIC_FLIST_DECLARE(items, 0, pa_xfree);
87
88 struct item_info {
89 enum {
90 PA_PSTREAM_ITEM_PACKET,
91 PA_PSTREAM_ITEM_MEMBLOCK,
92 PA_PSTREAM_ITEM_SHMRELEASE,
93 PA_PSTREAM_ITEM_SHMREVOKE
94 } type;
95
96 /* packet info */
97 pa_packet *packet;
98 #ifdef HAVE_CREDS
99 pa_bool_t with_creds;
100 pa_creds creds;
101 #endif
102
103 /* memblock info */
104 pa_memchunk chunk;
105 uint32_t channel;
106 int64_t offset;
107 pa_seek_mode_t seek_mode;
108
109 /* release/revoke info */
110 uint32_t block_id;
111 };
112
113 struct pa_pstream {
114 PA_REFCNT_DECLARE;
115
116 pa_mainloop_api *mainloop;
117 pa_defer_event *defer_event;
118 pa_iochannel *io;
119
120 pa_queue *send_queue;
121
122 pa_bool_t dead;
123
124 struct {
125 pa_pstream_descriptor descriptor;
126 struct item_info* current;
127 uint32_t shm_info[PA_PSTREAM_SHM_MAX];
128 void *data;
129 size_t index;
130 pa_memchunk memchunk;
131 } write;
132
133 struct {
134 pa_pstream_descriptor descriptor;
135 pa_memblock *memblock;
136 pa_packet *packet;
137 uint32_t shm_info[PA_PSTREAM_SHM_MAX];
138 void *data;
139 size_t index;
140 } read;
141
142 pa_bool_t use_shm;
143 pa_memimport *import;
144 pa_memexport *export;
145
146 pa_pstream_packet_cb_t recieve_packet_callback;
147 void *recieve_packet_callback_userdata;
148
149 pa_pstream_memblock_cb_t recieve_memblock_callback;
150 void *recieve_memblock_callback_userdata;
151
152 pa_pstream_notify_cb_t drain_callback;
153 void *drain_callback_userdata;
154
155 pa_pstream_notify_cb_t die_callback;
156 void *die_callback_userdata;
157
158 pa_pstream_block_id_cb_t revoke_callback;
159 void *revoke_callback_userdata;
160
161 pa_pstream_block_id_cb_t release_callback;
162 void *release_callback_userdata;
163
164 pa_mempool *mempool;
165
166 #ifdef HAVE_CREDS
167 pa_creds read_creds, write_creds;
168 pa_bool_t read_creds_valid, send_creds_now;
169 #endif
170 };
171
172 static int do_write(pa_pstream *p);
173 static int do_read(pa_pstream *p);
174
175 static void do_something(pa_pstream *p) {
176 pa_assert(p);
177 pa_assert(PA_REFCNT_VALUE(p) > 0);
178
179 pa_pstream_ref(p);
180
181 p->mainloop->defer_enable(p->defer_event, 0);
182
183 if (!p->dead && pa_iochannel_is_readable(p->io)) {
184 if (do_read(p) < 0)
185 goto fail;
186 } else if (!p->dead && pa_iochannel_is_hungup(p->io))
187 goto fail;
188
189 if (!p->dead && pa_iochannel_is_writable(p->io)) {
190 if (do_write(p) < 0)
191 goto fail;
192 }
193
194 pa_pstream_unref(p);
195 return;
196
197 fail:
198
199 if (p->die_callback)
200 p->die_callback(p, p->die_callback_userdata);
201
202 pa_pstream_unlink(p);
203 pa_pstream_unref(p);
204 }
205
206 static void io_callback(pa_iochannel*io, void *userdata) {
207 pa_pstream *p = userdata;
208
209 pa_assert(p);
210 pa_assert(PA_REFCNT_VALUE(p) > 0);
211 pa_assert(p->io == io);
212
213 do_something(p);
214 }
215
216 static void defer_callback(pa_mainloop_api *m, pa_defer_event *e, void*userdata) {
217 pa_pstream *p = userdata;
218
219 pa_assert(p);
220 pa_assert(PA_REFCNT_VALUE(p) > 0);
221 pa_assert(p->defer_event == e);
222 pa_assert(p->mainloop == m);
223
224 do_something(p);
225 }
226
227 static void memimport_release_cb(pa_memimport *i, uint32_t block_id, void *userdata);
228
229 pa_pstream *pa_pstream_new(pa_mainloop_api *m, pa_iochannel *io, pa_mempool *pool) {
230 pa_pstream *p;
231
232 pa_assert(m);
233 pa_assert(io);
234 pa_assert(pool);
235
236 p = pa_xnew(pa_pstream, 1);
237 PA_REFCNT_INIT(p);
238 p->io = io;
239 pa_iochannel_set_callback(io, io_callback, p);
240 p->dead = FALSE;
241
242 p->mainloop = m;
243 p->defer_event = m->defer_new(m, defer_callback, p);
244 m->defer_enable(p->defer_event, 0);
245
246 p->send_queue = pa_queue_new();
247
248 p->write.current = NULL;
249 p->write.index = 0;
250 pa_memchunk_reset(&p->write.memchunk);
251 p->read.memblock = NULL;
252 p->read.packet = NULL;
253 p->read.index = 0;
254
255 p->recieve_packet_callback = NULL;
256 p->recieve_packet_callback_userdata = NULL;
257 p->recieve_memblock_callback = NULL;
258 p->recieve_memblock_callback_userdata = NULL;
259 p->drain_callback = NULL;
260 p->drain_callback_userdata = NULL;
261 p->die_callback = NULL;
262 p->die_callback_userdata = NULL;
263 p->revoke_callback = NULL;
264 p->revoke_callback_userdata = NULL;
265 p->release_callback = NULL;
266 p->release_callback_userdata = NULL;
267
268 p->mempool = pool;
269
270 p->use_shm = FALSE;
271 p->export = NULL;
272
273 /* We do importing unconditionally */
274 p->import = pa_memimport_new(p->mempool, memimport_release_cb, p);
275
276 pa_iochannel_socket_set_rcvbuf(io, pa_mempool_block_size_max(p->mempool));
277 pa_iochannel_socket_set_sndbuf(io, pa_mempool_block_size_max(p->mempool));
278
279 #ifdef HAVE_CREDS
280 p->send_creds_now = FALSE;
281 p->read_creds_valid = FALSE;
282 #endif
283 return p;
284 }
285
286 static void item_free(void *item, void *q) {
287 struct item_info *i = item;
288 pa_assert(i);
289
290 if (i->type == PA_PSTREAM_ITEM_MEMBLOCK) {
291 pa_assert(i->chunk.memblock);
292 pa_memblock_unref(i->chunk.memblock);
293 } else if (i->type == PA_PSTREAM_ITEM_PACKET) {
294 pa_assert(i->packet);
295 pa_packet_unref(i->packet);
296 }
297
298 if (pa_flist_push(PA_STATIC_FLIST_GET(items), i) < 0)
299 pa_xfree(i);
300 }
301
302 static void pstream_free(pa_pstream *p) {
303 pa_assert(p);
304
305 pa_pstream_unlink(p);
306
307 pa_queue_free(p->send_queue, item_free, NULL);
308
309 if (p->write.current)
310 item_free(p->write.current, NULL);
311
312 if (p->write.memchunk.memblock)
313 pa_memblock_unref(p->write.memchunk.memblock);
314
315 if (p->read.memblock)
316 pa_memblock_unref(p->read.memblock);
317
318 if (p->read.packet)
319 pa_packet_unref(p->read.packet);
320
321 pa_xfree(p);
322 }
323
324 void pa_pstream_send_packet(pa_pstream*p, pa_packet *packet, const pa_creds *creds) {
325 struct item_info *i;
326
327 pa_assert(p);
328 pa_assert(PA_REFCNT_VALUE(p) > 0);
329 pa_assert(packet);
330
331 if (p->dead)
332 return;
333
334 if (!(i = pa_flist_pop(PA_STATIC_FLIST_GET(items))))
335 i = pa_xnew(struct item_info, 1);
336
337 i->type = PA_PSTREAM_ITEM_PACKET;
338 i->packet = pa_packet_ref(packet);
339
340 #ifdef HAVE_CREDS
341 if ((i->with_creds = !!creds))
342 i->creds = *creds;
343 #endif
344
345 pa_queue_push(p->send_queue, i);
346
347 p->mainloop->defer_enable(p->defer_event, 1);
348 }
349
350 void pa_pstream_send_memblock(pa_pstream*p, uint32_t channel, int64_t offset, pa_seek_mode_t seek_mode, const pa_memchunk *chunk) {
351 size_t length, idx;
352 size_t bsm;
353
354 pa_assert(p);
355 pa_assert(PA_REFCNT_VALUE(p) > 0);
356 pa_assert(channel != (uint32_t) -1);
357 pa_assert(chunk);
358
359 if (p->dead)
360 return;
361
362 idx = 0;
363 length = chunk->length;
364
365 bsm = pa_mempool_block_size_max(p->mempool);
366
367 while (length > 0) {
368 struct item_info *i;
369 size_t n;
370
371 if (!(i = pa_flist_pop(PA_STATIC_FLIST_GET(items))))
372 i = pa_xnew(struct item_info, 1);
373 i->type = PA_PSTREAM_ITEM_MEMBLOCK;
374
375 n = PA_MIN(length, bsm);
376 i->chunk.index = chunk->index + idx;
377 i->chunk.length = n;
378 i->chunk.memblock = pa_memblock_ref(chunk->memblock);
379
380 i->channel = channel;
381 i->offset = offset;
382 i->seek_mode = seek_mode;
383 #ifdef HAVE_CREDS
384 i->with_creds = FALSE;
385 #endif
386
387 pa_queue_push(p->send_queue, i);
388
389 idx += n;
390 length -= n;
391 }
392
393 p->mainloop->defer_enable(p->defer_event, 1);
394 }
395
396 void pa_pstream_send_release(pa_pstream *p, uint32_t block_id) {
397 struct item_info *item;
398 pa_assert(p);
399 pa_assert(PA_REFCNT_VALUE(p) > 0);
400
401 if (p->dead)
402 return;
403
404 /* pa_log("Releasing block %u", block_id); */
405
406 if (!(item = pa_flist_pop(PA_STATIC_FLIST_GET(items))))
407 item = pa_xnew(struct item_info, 1);
408 item->type = PA_PSTREAM_ITEM_SHMRELEASE;
409 item->block_id = block_id;
410 #ifdef HAVE_CREDS
411 item->with_creds = FALSE;
412 #endif
413
414 pa_queue_push(p->send_queue, item);
415 p->mainloop->defer_enable(p->defer_event, 1);
416 }
417
418 /* might be called from thread context */
419 static void memimport_release_cb(pa_memimport *i, uint32_t block_id, void *userdata) {
420 pa_pstream *p = userdata;
421
422 pa_assert(p);
423 pa_assert(PA_REFCNT_VALUE(p) > 0);
424
425 if (p->dead)
426 return;
427
428 if (p->release_callback)
429 p->release_callback(p, block_id, p->release_callback_userdata);
430 else
431 pa_pstream_send_release(p, block_id);
432 }
433
434 void pa_pstream_send_revoke(pa_pstream *p, uint32_t block_id) {
435 struct item_info *item;
436 pa_assert(p);
437 pa_assert(PA_REFCNT_VALUE(p) > 0);
438
439 if (p->dead)
440 return;
441 /* pa_log("Revoking block %u", block_id); */
442
443 if (!(item = pa_flist_pop(PA_STATIC_FLIST_GET(items))))
444 item = pa_xnew(struct item_info, 1);
445 item->type = PA_PSTREAM_ITEM_SHMREVOKE;
446 item->block_id = block_id;
447 #ifdef HAVE_CREDS
448 item->with_creds = FALSE;
449 #endif
450
451 pa_queue_push(p->send_queue, item);
452 p->mainloop->defer_enable(p->defer_event, 1);
453 }
454
455 /* might be called from thread context */
456 static void memexport_revoke_cb(pa_memexport *e, uint32_t block_id, void *userdata) {
457 pa_pstream *p = userdata;
458
459 pa_assert(p);
460 pa_assert(PA_REFCNT_VALUE(p) > 0);
461
462 if (p->revoke_callback)
463 p->revoke_callback(p, block_id, p->revoke_callback_userdata);
464 else
465 pa_pstream_send_revoke(p, block_id);
466 }
467
468 static void prepare_next_write_item(pa_pstream *p) {
469 pa_assert(p);
470 pa_assert(PA_REFCNT_VALUE(p) > 0);
471
472 p->write.current = pa_queue_pop(p->send_queue);
473
474 if (!p->write.current)
475 return;
476
477 p->write.index = 0;
478 p->write.data = NULL;
479 pa_memchunk_reset(&p->write.memchunk);
480
481 p->write.descriptor[PA_PSTREAM_DESCRIPTOR_LENGTH] = 0;
482 p->write.descriptor[PA_PSTREAM_DESCRIPTOR_CHANNEL] = htonl((uint32_t) -1);
483 p->write.descriptor[PA_PSTREAM_DESCRIPTOR_OFFSET_HI] = 0;
484 p->write.descriptor[PA_PSTREAM_DESCRIPTOR_OFFSET_LO] = 0;
485 p->write.descriptor[PA_PSTREAM_DESCRIPTOR_FLAGS] = 0;
486
487 if (p->write.current->type == PA_PSTREAM_ITEM_PACKET) {
488
489 pa_assert(p->write.current->packet);
490 p->write.data = p->write.current->packet->data;
491 p->write.descriptor[PA_PSTREAM_DESCRIPTOR_LENGTH] = htonl((uint32_t) p->write.current->packet->length);
492
493 } else if (p->write.current->type == PA_PSTREAM_ITEM_SHMRELEASE) {
494
495 p->write.descriptor[PA_PSTREAM_DESCRIPTOR_FLAGS] = htonl(PA_FLAG_SHMRELEASE);
496 p->write.descriptor[PA_PSTREAM_DESCRIPTOR_OFFSET_HI] = htonl(p->write.current->block_id);
497
498 } else if (p->write.current->type == PA_PSTREAM_ITEM_SHMREVOKE) {
499
500 p->write.descriptor[PA_PSTREAM_DESCRIPTOR_FLAGS] = htonl(PA_FLAG_SHMREVOKE);
501 p->write.descriptor[PA_PSTREAM_DESCRIPTOR_OFFSET_HI] = htonl(p->write.current->block_id);
502
503 } else {
504 uint32_t flags;
505 pa_bool_t send_payload = TRUE;
506
507 pa_assert(p->write.current->type == PA_PSTREAM_ITEM_MEMBLOCK);
508 pa_assert(p->write.current->chunk.memblock);
509
510 p->write.descriptor[PA_PSTREAM_DESCRIPTOR_CHANNEL] = htonl(p->write.current->channel);
511 p->write.descriptor[PA_PSTREAM_DESCRIPTOR_OFFSET_HI] = htonl((uint32_t) (((uint64_t) p->write.current->offset) >> 32));
512 p->write.descriptor[PA_PSTREAM_DESCRIPTOR_OFFSET_LO] = htonl((uint32_t) ((uint64_t) p->write.current->offset));
513
514 flags = (uint32_t) (p->write.current->seek_mode & PA_FLAG_SEEKMASK);
515
516 if (p->use_shm) {
517 uint32_t block_id, shm_id;
518 size_t offset, length;
519
520 pa_assert(p->export);
521
522 if (pa_memexport_put(p->export,
523 p->write.current->chunk.memblock,
524 &block_id,
525 &shm_id,
526 &offset,
527 &length) >= 0) {
528
529 flags |= PA_FLAG_SHMDATA;
530 send_payload = FALSE;
531
532 p->write.shm_info[PA_PSTREAM_SHM_BLOCKID] = htonl(block_id);
533 p->write.shm_info[PA_PSTREAM_SHM_SHMID] = htonl(shm_id);
534 p->write.shm_info[PA_PSTREAM_SHM_INDEX] = htonl((uint32_t) (offset + p->write.current->chunk.index));
535 p->write.shm_info[PA_PSTREAM_SHM_LENGTH] = htonl((uint32_t) p->write.current->chunk.length);
536
537 p->write.descriptor[PA_PSTREAM_DESCRIPTOR_LENGTH] = htonl(sizeof(p->write.shm_info));
538 p->write.data = p->write.shm_info;
539 }
540 /* else */
541 /* pa_log_warn("Failed to export memory block."); */
542 }
543
544 if (send_payload) {
545 p->write.descriptor[PA_PSTREAM_DESCRIPTOR_LENGTH] = htonl((uint32_t) p->write.current->chunk.length);
546 p->write.memchunk = p->write.current->chunk;
547 pa_memblock_ref(p->write.memchunk.memblock);
548 p->write.data = NULL;
549 }
550
551 p->write.descriptor[PA_PSTREAM_DESCRIPTOR_FLAGS] = htonl(flags);
552 }
553
554 #ifdef HAVE_CREDS
555 if ((p->send_creds_now = p->write.current->with_creds))
556 p->write_creds = p->write.current->creds;
557 #endif
558 }
559
560 static int do_write(pa_pstream *p) {
561 void *d;
562 size_t l;
563 ssize_t r;
564 pa_memblock *release_memblock = NULL;
565
566 pa_assert(p);
567 pa_assert(PA_REFCNT_VALUE(p) > 0);
568
569 if (!p->write.current)
570 prepare_next_write_item(p);
571
572 if (!p->write.current)
573 return 0;
574
575 if (p->write.index < PA_PSTREAM_DESCRIPTOR_SIZE) {
576 d = (uint8_t*) p->write.descriptor + p->write.index;
577 l = PA_PSTREAM_DESCRIPTOR_SIZE - p->write.index;
578 } else {
579 pa_assert(p->write.data || p->write.memchunk.memblock);
580
581 if (p->write.data)
582 d = p->write.data;
583 else {
584 d = (uint8_t*) pa_memblock_acquire(p->write.memchunk.memblock) + p->write.memchunk.index;
585 release_memblock = p->write.memchunk.memblock;
586 }
587
588 d = (uint8_t*) d + p->write.index - PA_PSTREAM_DESCRIPTOR_SIZE;
589 l = ntohl(p->write.descriptor[PA_PSTREAM_DESCRIPTOR_LENGTH]) - (p->write.index - PA_PSTREAM_DESCRIPTOR_SIZE);
590 }
591
592 pa_assert(l > 0);
593
594 #ifdef HAVE_CREDS
595 if (p->send_creds_now) {
596
597 if ((r = pa_iochannel_write_with_creds(p->io, d, l, &p->write_creds)) < 0)
598 goto fail;
599
600 p->send_creds_now = FALSE;
601 } else
602 #endif
603
604 if ((r = pa_iochannel_write(p->io, d, l)) < 0)
605 goto fail;
606
607 if (release_memblock)
608 pa_memblock_release(release_memblock);
609
610 p->write.index += (size_t) r;
611
612 if (p->write.index >= PA_PSTREAM_DESCRIPTOR_SIZE + ntohl(p->write.descriptor[PA_PSTREAM_DESCRIPTOR_LENGTH])) {
613 pa_assert(p->write.current);
614 item_free(p->write.current, NULL);
615 p->write.current = NULL;
616
617 if (p->write.memchunk.memblock)
618 pa_memblock_unref(p->write.memchunk.memblock);
619
620 pa_memchunk_reset(&p->write.memchunk);
621
622 if (p->drain_callback && !pa_pstream_is_pending(p))
623 p->drain_callback(p, p->drain_callback_userdata);
624 }
625
626 return 0;
627
628 fail:
629
630 if (release_memblock)
631 pa_memblock_release(release_memblock);
632
633 return -1;
634 }
635
636 static int do_read(pa_pstream *p) {
637 void *d;
638 size_t l;
639 ssize_t r;
640 pa_memblock *release_memblock = NULL;
641 pa_assert(p);
642 pa_assert(PA_REFCNT_VALUE(p) > 0);
643
644 if (p->read.index < PA_PSTREAM_DESCRIPTOR_SIZE) {
645 d = (uint8_t*) p->read.descriptor + p->read.index;
646 l = PA_PSTREAM_DESCRIPTOR_SIZE - p->read.index;
647 } else {
648 pa_assert(p->read.data || p->read.memblock);
649
650 if (p->read.data)
651 d = p->read.data;
652 else {
653 d = pa_memblock_acquire(p->read.memblock);
654 release_memblock = p->read.memblock;
655 }
656
657 d = (uint8_t*) d + p->read.index - PA_PSTREAM_DESCRIPTOR_SIZE;
658 l = ntohl(p->read.descriptor[PA_PSTREAM_DESCRIPTOR_LENGTH]) - (p->read.index - PA_PSTREAM_DESCRIPTOR_SIZE);
659 }
660
661 #ifdef HAVE_CREDS
662 {
663 pa_bool_t b = 0;
664
665 if ((r = pa_iochannel_read_with_creds(p->io, d, l, &p->read_creds, &b)) <= 0)
666 goto fail;
667
668 p->read_creds_valid = p->read_creds_valid || b;
669 }
670 #else
671 if ((r = pa_iochannel_read(p->io, d, l)) <= 0)
672 goto fail;
673 #endif
674
675 if (release_memblock)
676 pa_memblock_release(release_memblock);
677
678 p->read.index += (size_t) r;
679
680 if (p->read.index == PA_PSTREAM_DESCRIPTOR_SIZE) {
681 uint32_t flags, length, channel;
682 /* Reading of frame descriptor complete */
683
684 flags = ntohl(p->read.descriptor[PA_PSTREAM_DESCRIPTOR_FLAGS]);
685
686 if (!p->use_shm && (flags & PA_FLAG_SHMMASK) != 0) {
687 pa_log_warn("Received SHM frame on a socket where SHM is disabled.");
688 return -1;
689 }
690
691 if (flags == PA_FLAG_SHMRELEASE) {
692
693 /* This is a SHM memblock release frame with no payload */
694
695 /* pa_log("Got release frame for %u", ntohl(p->read.descriptor[PA_PSTREAM_DESCRIPTOR_OFFSET_HI])); */
696
697 pa_assert(p->export);
698 pa_memexport_process_release(p->export, ntohl(p->read.descriptor[PA_PSTREAM_DESCRIPTOR_OFFSET_HI]));
699
700 goto frame_done;
701
702 } else if (flags == PA_FLAG_SHMREVOKE) {
703
704 /* This is a SHM memblock revoke frame with no payload */
705
706 /* pa_log("Got revoke frame for %u", ntohl(p->read.descriptor[PA_PSTREAM_DESCRIPTOR_OFFSET_HI])); */
707
708 pa_assert(p->import);
709 pa_memimport_process_revoke(p->import, ntohl(p->read.descriptor[PA_PSTREAM_DESCRIPTOR_OFFSET_HI]));
710
711 goto frame_done;
712 }
713
714 length = ntohl(p->read.descriptor[PA_PSTREAM_DESCRIPTOR_LENGTH]);
715
716 if (length > FRAME_SIZE_MAX_ALLOW || length <= 0) {
717 pa_log_warn("Received invalid frame size: %lu", (unsigned long) length);
718 return -1;
719 }
720
721 pa_assert(!p->read.packet && !p->read.memblock);
722
723 channel = ntohl(p->read.descriptor[PA_PSTREAM_DESCRIPTOR_CHANNEL]);
724
725 if (channel == (uint32_t) -1) {
726
727 if (flags != 0) {
728 pa_log_warn("Received packet frame with invalid flags value.");
729 return -1;
730 }
731
732 /* Frame is a packet frame */
733 p->read.packet = pa_packet_new(length);
734 p->read.data = p->read.packet->data;
735
736 } else {
737
738 if ((flags & PA_FLAG_SEEKMASK) > PA_SEEK_RELATIVE_END) {
739 pa_log_warn("Received memblock frame with invalid seek mode.");
740 return -1;
741 }
742
743 if ((flags & PA_FLAG_SHMMASK) == PA_FLAG_SHMDATA) {
744
745 if (length != sizeof(p->read.shm_info)) {
746 pa_log_warn("Received SHM memblock frame with Invalid frame length.");
747 return -1;
748 }
749
750 /* Frame is a memblock frame referencing an SHM memblock */
751 p->read.data = p->read.shm_info;
752
753 } else if ((flags & PA_FLAG_SHMMASK) == 0) {
754
755 /* Frame is a memblock frame */
756
757 p->read.memblock = pa_memblock_new(p->mempool, length);
758 p->read.data = NULL;
759 } else {
760
761 pa_log_warn("Received memblock frame with invalid flags value.");
762 return -1;
763 }
764 }
765
766 } else if (p->read.index > PA_PSTREAM_DESCRIPTOR_SIZE) {
767 /* Frame payload available */
768
769 if (p->read.memblock && p->recieve_memblock_callback) {
770
771 /* Is this memblock data? Than pass it to the user */
772 l = (p->read.index - (size_t) r) < PA_PSTREAM_DESCRIPTOR_SIZE ? (size_t) (p->read.index - PA_PSTREAM_DESCRIPTOR_SIZE) : (size_t) r;
773
774 if (l > 0) {
775 pa_memchunk chunk;
776
777 chunk.memblock = p->read.memblock;
778 chunk.index = p->read.index - PA_PSTREAM_DESCRIPTOR_SIZE - l;
779 chunk.length = l;
780
781 if (p->recieve_memblock_callback) {
782 int64_t offset;
783
784 offset = (int64_t) (
785 (((uint64_t) ntohl(p->read.descriptor[PA_PSTREAM_DESCRIPTOR_OFFSET_HI])) << 32) |
786 (((uint64_t) ntohl(p->read.descriptor[PA_PSTREAM_DESCRIPTOR_OFFSET_LO]))));
787
788 p->recieve_memblock_callback(
789 p,
790 ntohl(p->read.descriptor[PA_PSTREAM_DESCRIPTOR_CHANNEL]),
791 offset,
792 ntohl(p->read.descriptor[PA_PSTREAM_DESCRIPTOR_FLAGS]) & PA_FLAG_SEEKMASK,
793 &chunk,
794 p->recieve_memblock_callback_userdata);
795 }
796
797 /* Drop seek info for following callbacks */
798 p->read.descriptor[PA_PSTREAM_DESCRIPTOR_FLAGS] =
799 p->read.descriptor[PA_PSTREAM_DESCRIPTOR_OFFSET_HI] =
800 p->read.descriptor[PA_PSTREAM_DESCRIPTOR_OFFSET_LO] = 0;
801 }
802 }
803
804 /* Frame complete */
805 if (p->read.index >= ntohl(p->read.descriptor[PA_PSTREAM_DESCRIPTOR_LENGTH]) + PA_PSTREAM_DESCRIPTOR_SIZE) {
806
807 if (p->read.memblock) {
808
809 /* This was a memblock frame. We can unref the memblock now */
810 pa_memblock_unref(p->read.memblock);
811
812 } else if (p->read.packet) {
813
814 if (p->recieve_packet_callback)
815 #ifdef HAVE_CREDS
816 p->recieve_packet_callback(p, p->read.packet, p->read_creds_valid ? &p->read_creds : NULL, p->recieve_packet_callback_userdata);
817 #else
818 p->recieve_packet_callback(p, p->read.packet, NULL, p->recieve_packet_callback_userdata);
819 #endif
820
821 pa_packet_unref(p->read.packet);
822 } else {
823 pa_memblock *b;
824
825 pa_assert((ntohl(p->read.descriptor[PA_PSTREAM_DESCRIPTOR_FLAGS]) & PA_FLAG_SHMMASK) == PA_FLAG_SHMDATA);
826
827 pa_assert(p->import);
828
829 if (!(b = pa_memimport_get(p->import,
830 ntohl(p->read.shm_info[PA_PSTREAM_SHM_BLOCKID]),
831 ntohl(p->read.shm_info[PA_PSTREAM_SHM_SHMID]),
832 ntohl(p->read.shm_info[PA_PSTREAM_SHM_INDEX]),
833 ntohl(p->read.shm_info[PA_PSTREAM_SHM_LENGTH])))) {
834
835 if (pa_log_ratelimit(PA_LOG_DEBUG))
836 pa_log_debug("Failed to import memory block.");
837 }
838
839 if (p->recieve_memblock_callback) {
840 int64_t offset;
841 pa_memchunk chunk;
842
843 chunk.memblock = b;
844 chunk.index = 0;
845 chunk.length = b ? pa_memblock_get_length(b) : ntohl(p->read.shm_info[PA_PSTREAM_SHM_LENGTH]);
846
847 offset = (int64_t) (
848 (((uint64_t) ntohl(p->read.descriptor[PA_PSTREAM_DESCRIPTOR_OFFSET_HI])) << 32) |
849 (((uint64_t) ntohl(p->read.descriptor[PA_PSTREAM_DESCRIPTOR_OFFSET_LO]))));
850
851 p->recieve_memblock_callback(
852 p,
853 ntohl(p->read.descriptor[PA_PSTREAM_DESCRIPTOR_CHANNEL]),
854 offset,
855 ntohl(p->read.descriptor[PA_PSTREAM_DESCRIPTOR_FLAGS]) & PA_FLAG_SEEKMASK,
856 &chunk,
857 p->recieve_memblock_callback_userdata);
858 }
859
860 if (b)
861 pa_memblock_unref(b);
862 }
863
864 goto frame_done;
865 }
866 }
867
868 return 0;
869
870 frame_done:
871 p->read.memblock = NULL;
872 p->read.packet = NULL;
873 p->read.index = 0;
874 p->read.data = NULL;
875
876 #ifdef HAVE_CREDS
877 p->read_creds_valid = FALSE;
878 #endif
879
880 return 0;
881
882 fail:
883 if (release_memblock)
884 pa_memblock_release(release_memblock);
885
886 return -1;
887 }
888
889 void pa_pstream_set_die_callback(pa_pstream *p, pa_pstream_notify_cb_t cb, void *userdata) {
890 pa_assert(p);
891 pa_assert(PA_REFCNT_VALUE(p) > 0);
892
893 p->die_callback = cb;
894 p->die_callback_userdata = userdata;
895 }
896
897 void pa_pstream_set_drain_callback(pa_pstream *p, pa_pstream_notify_cb_t cb, void *userdata) {
898 pa_assert(p);
899 pa_assert(PA_REFCNT_VALUE(p) > 0);
900
901 p->drain_callback = cb;
902 p->drain_callback_userdata = userdata;
903 }
904
905 void pa_pstream_set_recieve_packet_callback(pa_pstream *p, pa_pstream_packet_cb_t cb, void *userdata) {
906 pa_assert(p);
907 pa_assert(PA_REFCNT_VALUE(p) > 0);
908
909 p->recieve_packet_callback = cb;
910 p->recieve_packet_callback_userdata = userdata;
911 }
912
913 void pa_pstream_set_recieve_memblock_callback(pa_pstream *p, pa_pstream_memblock_cb_t cb, void *userdata) {
914 pa_assert(p);
915 pa_assert(PA_REFCNT_VALUE(p) > 0);
916
917 p->recieve_memblock_callback = cb;
918 p->recieve_memblock_callback_userdata = userdata;
919 }
920
921 void pa_pstream_set_release_callback(pa_pstream *p, pa_pstream_block_id_cb_t cb, void *userdata) {
922 pa_assert(p);
923 pa_assert(PA_REFCNT_VALUE(p) > 0);
924
925 p->release_callback = cb;
926 p->release_callback_userdata = userdata;
927 }
928
929 void pa_pstream_set_revoke_callback(pa_pstream *p, pa_pstream_block_id_cb_t cb, void *userdata) {
930 pa_assert(p);
931 pa_assert(PA_REFCNT_VALUE(p) > 0);
932
933 p->release_callback = cb;
934 p->release_callback_userdata = userdata;
935 }
936
937 pa_bool_t pa_pstream_is_pending(pa_pstream *p) {
938 pa_bool_t b;
939
940 pa_assert(p);
941 pa_assert(PA_REFCNT_VALUE(p) > 0);
942
943 if (p->dead)
944 b = FALSE;
945 else
946 b = p->write.current || !pa_queue_isempty(p->send_queue);
947
948 return b;
949 }
950
951 void pa_pstream_unref(pa_pstream*p) {
952 pa_assert(p);
953 pa_assert(PA_REFCNT_VALUE(p) > 0);
954
955 if (PA_REFCNT_DEC(p) <= 0)
956 pstream_free(p);
957 }
958
959 pa_pstream* pa_pstream_ref(pa_pstream*p) {
960 pa_assert(p);
961 pa_assert(PA_REFCNT_VALUE(p) > 0);
962
963 PA_REFCNT_INC(p);
964 return p;
965 }
966
967 void pa_pstream_unlink(pa_pstream *p) {
968 pa_assert(p);
969
970 if (p->dead)
971 return;
972
973 p->dead = TRUE;
974
975 if (p->import) {
976 pa_memimport_free(p->import);
977 p->import = NULL;
978 }
979
980 if (p->export) {
981 pa_memexport_free(p->export);
982 p->export = NULL;
983 }
984
985 if (p->io) {
986 pa_iochannel_free(p->io);
987 p->io = NULL;
988 }
989
990 if (p->defer_event) {
991 p->mainloop->defer_free(p->defer_event);
992 p->defer_event = NULL;
993 }
994
995 p->die_callback = NULL;
996 p->drain_callback = NULL;
997 p->recieve_packet_callback = NULL;
998 p->recieve_memblock_callback = NULL;
999 }
1000
1001 void pa_pstream_enable_shm(pa_pstream *p, pa_bool_t enable) {
1002 pa_assert(p);
1003 pa_assert(PA_REFCNT_VALUE(p) > 0);
1004
1005 p->use_shm = enable;
1006
1007 if (enable) {
1008
1009 if (!p->export)
1010 p->export = pa_memexport_new(p->mempool, memexport_revoke_cb, p);
1011
1012 } else {
1013
1014 if (p->export) {
1015 pa_memexport_free(p->export);
1016 p->export = NULL;
1017 }
1018 }
1019 }
1020
1021 pa_bool_t pa_pstream_get_shm(pa_pstream *p) {
1022 pa_assert(p);
1023 pa_assert(PA_REFCNT_VALUE(p) > 0);
1024
1025 return p->use_shm;
1026 }
A mirror of the PulseAudio repo