3 This file is part of PulseAudio.
5 Copyright 2006 Lennart Poettering
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 published
9 by the Free Software Foundation; either version 2.1 of the License,
10 or (at your option) any later version.
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 General Public License for more details.
17 You should have received a copy of the GNU Lesser General Public License
18 along with PulseAudio; if not, write to the Free Software
19 Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307
28 #include <sys/socket.h>
29 #include <netinet/in.h>
30 #include <arpa/inet.h>
36 #include <pulse/rtclock.h>
37 #include <pulse/timeval.h>
38 #include <pulse/xmalloc.h>
40 #include <pulsecore/core-error.h>
41 #include <pulsecore/module.h>
42 #include <pulsecore/llist.h>
43 #include <pulsecore/sink.h>
44 #include <pulsecore/sink-input.h>
45 #include <pulsecore/memblockq.h>
46 #include <pulsecore/log.h>
47 #include <pulsecore/core-rtclock.h>
48 #include <pulsecore/core-util.h>
49 #include <pulsecore/modargs.h>
50 #include <pulsecore/namereg.h>
51 #include <pulsecore/sample-util.h>
52 #include <pulsecore/macro.h>
53 #include <pulsecore/atomic.h>
54 #include <pulsecore/atomic.h>
55 #include <pulsecore/time-smoother.h>
56 #include <pulsecore/socket-util.h>
57 #include <pulsecore/once.h>
59 #include "module-rtp-recv-symdef.h"
65 PA_MODULE_AUTHOR("Lennart Poettering");
66 PA_MODULE_DESCRIPTION("Receive data from a network via RTP/SAP/SDP");
67 PA_MODULE_VERSION(PACKAGE_VERSION
);
68 PA_MODULE_LOAD_ONCE(FALSE
);
70 "sink=<name of the sink> "
71 "sap_address=<multicast address to listen on> "
75 #define DEFAULT_SAP_ADDRESS "224.0.0.56"
76 #define MEMBLOCKQ_MAXLENGTH (1024*1024*40)
77 #define MAX_SESSIONS 16
78 #define DEATH_TIMEOUT 20
79 #define RATE_UPDATE_INTERVAL (5*PA_USEC_PER_SEC)
80 #define LATENCY_USEC (500*PA_USEC_PER_MSEC)
82 static const char* const valid_modargs
[] = {
89 struct userdata
*userdata
;
90 PA_LLIST_FIELDS(struct session
);
92 pa_sink_input
*sink_input
;
93 pa_memblockq
*memblockq
;
95 pa_bool_t first_packet
;
99 struct pa_sdp_info sdp_info
;
101 pa_rtp_context rtp_context
;
103 pa_rtpoll_item
*rtpoll_item
;
105 pa_atomic_t timestamp
;
107 pa_smoother
*smoother
;
108 pa_usec_t intended_latency
;
109 pa_usec_t sink_latency
;
111 pa_usec_t last_rate_update
;
112 pa_usec_t last_latency
;
113 double estimated_rate
;
114 double avg_estimated_rate
;
121 pa_sap_context sap_context
;
122 pa_io_event
* sap_event
;
124 pa_time_event
*check_death_event
;
128 PA_LLIST_HEAD(struct session
, sessions
);
129 pa_hashmap
*by_origin
;
133 static void session_free(struct session
*s
);
135 /* Called from I/O thread context */
136 static int sink_input_process_msg(pa_msgobject
*o
, int code
, void *data
, int64_t offset
, pa_memchunk
*chunk
) {
137 struct session
*s
= PA_SINK_INPUT(o
)->userdata
;
140 case PA_SINK_INPUT_MESSAGE_GET_LATENCY
:
141 *((pa_usec_t
*) data
) = pa_bytes_to_usec(pa_memblockq_get_length(s
->memblockq
), &s
->sink_input
->sample_spec
);
143 /* Fall through, the default handler will add in the extra
144 * latency added by the resampler */
148 return pa_sink_input_process_msg(o
, code
, data
, offset
, chunk
);
151 /* Called from I/O thread context */
152 static int sink_input_pop_cb(pa_sink_input
*i
, size_t length
, pa_memchunk
*chunk
) {
154 pa_sink_input_assert_ref(i
);
155 pa_assert_se(s
= i
->userdata
);
157 if (pa_memblockq_peek(s
->memblockq
, chunk
) < 0)
160 pa_memblockq_drop(s
->memblockq
, chunk
->length
);
165 /* Called from I/O thread context */
166 static void sink_input_process_rewind_cb(pa_sink_input
*i
, size_t nbytes
) {
169 pa_sink_input_assert_ref(i
);
170 pa_assert_se(s
= i
->userdata
);
172 pa_memblockq_rewind(s
->memblockq
, nbytes
);
175 /* Called from I/O thread context */
176 static void sink_input_update_max_rewind_cb(pa_sink_input
*i
, size_t nbytes
) {
179 pa_sink_input_assert_ref(i
);
180 pa_assert_se(s
= i
->userdata
);
182 pa_memblockq_set_maxrewind(s
->memblockq
, nbytes
);
185 /* Called from main context */
186 static void sink_input_kill(pa_sink_input
* i
) {
188 pa_sink_input_assert_ref(i
);
189 pa_assert_se(s
= i
->userdata
);
194 /* Called from IO context */
195 static void sink_input_suspend_within_thread(pa_sink_input
* i
, pa_bool_t b
) {
197 pa_sink_input_assert_ref(i
);
198 pa_assert_se(s
= i
->userdata
);
201 pa_smoother_pause(s
->smoother
, pa_rtclock_now());
202 pa_memblockq_flush_read(s
->memblockq
);
204 s
->first_packet
= FALSE
;
207 /* Called from I/O thread context */
208 static int rtpoll_work_cb(pa_rtpoll_item
*i
) {
211 struct timeval now
= { 0, 0 };
215 pa_assert_se(s
= pa_rtpoll_item_get_userdata(i
));
217 p
= pa_rtpoll_item_get_pollfd(i
, NULL
);
219 if (p
->revents
& (POLLERR
|POLLNVAL
|POLLHUP
|POLLOUT
)) {
220 pa_log("poll() signalled bad revents.");
224 if ((p
->revents
& POLLIN
) == 0)
229 if (pa_rtp_recv(&s
->rtp_context
, &chunk
, s
->userdata
->module
->core
->mempool
, &now
) < 0)
232 if (s
->sdp_info
.payload
!= s
->rtp_context
.payload
||
233 !PA_SINK_IS_OPENED(s
->sink_input
->sink
->thread_info
.state
)) {
234 pa_memblock_unref(chunk
.memblock
);
238 if (!s
->first_packet
) {
239 s
->first_packet
= TRUE
;
241 s
->ssrc
= s
->rtp_context
.ssrc
;
242 s
->offset
= s
->rtp_context
.timestamp
;
244 if (s
->ssrc
== s
->userdata
->module
->core
->cookie
)
245 pa_log_warn("Detected RTP packet loop!");
247 if (s
->ssrc
!= s
->rtp_context
.ssrc
) {
248 pa_memblock_unref(chunk
.memblock
);
253 /* Check whether there was a timestamp overflow */
254 k
= (int64_t) s
->rtp_context
.timestamp
- (int64_t) s
->offset
;
255 j
= (int64_t) 0x100000000LL
- (int64_t) s
->offset
+ (int64_t) s
->rtp_context
.timestamp
;
257 if ((k
< 0 ? -k
: k
) < (j
< 0 ? -j
: j
))
262 pa_memblockq_seek(s
->memblockq
, delta
* (int64_t) s
->rtp_context
.frame_size
, PA_SEEK_RELATIVE
, TRUE
);
264 if (now
.tv_sec
== 0) {
266 pa_log_warn("Using artificial time instead of timestamp");
268 pa_rtclock_get(&now
);
270 pa_rtclock_from_wallclock(&now
);
272 pa_smoother_put(s
->smoother
, pa_timeval_load(&now
), pa_bytes_to_usec((uint64_t) pa_memblockq_get_write_index(s
->memblockq
), &s
->sink_input
->sample_spec
));
274 /* Tell the smoother that we are rolling now, in case it is still paused */
275 pa_smoother_resume(s
->smoother
, pa_timeval_load(&now
), TRUE
);
277 if (pa_memblockq_push(s
->memblockq
, &chunk
) < 0) {
278 pa_log_warn("Queue overrun");
279 pa_memblockq_seek(s
->memblockq
, (int64_t) chunk
.length
, PA_SEEK_RELATIVE
, TRUE
);
282 /* pa_log("blocks in q: %u", pa_memblockq_get_nblocks(s->memblockq)); */
284 pa_memblock_unref(chunk
.memblock
);
286 /* The next timestamp we expect */
287 s
->offset
= s
->rtp_context
.timestamp
+ (uint32_t) (chunk
.length
/ s
->rtp_context
.frame_size
);
289 pa_atomic_store(&s
->timestamp
, (int) now
.tv_sec
);
291 if (s
->last_rate_update
+ RATE_UPDATE_INTERVAL
< pa_timeval_load(&now
)) {
292 pa_usec_t wi
, ri
, render_delay
, sink_delay
= 0, latency
;
293 uint32_t base_rate
= s
->sink_input
->sink
->sample_spec
.rate
;
294 uint32_t current_rate
= s
->sink_input
->sample_spec
.rate
;
296 double estimated_rate
, alpha
= 0.02;
298 pa_log_debug("Updating sample rate");
300 wi
= pa_smoother_get(s
->smoother
, pa_timeval_load(&now
));
301 ri
= pa_bytes_to_usec((uint64_t) pa_memblockq_get_read_index(s
->memblockq
), &s
->sink_input
->sample_spec
);
303 pa_log_debug("wi=%lu ri=%lu", (unsigned long) wi
, (unsigned long) ri
);
305 sink_delay
= pa_sink_get_latency_within_thread(s
->sink_input
->sink
);
306 render_delay
= pa_bytes_to_usec(pa_memblockq_get_length(s
->sink_input
->thread_info
.render_memblockq
), &s
->sink_input
->sink
->sample_spec
);
308 if (ri
> render_delay
+sink_delay
)
309 ri
-= render_delay
+sink_delay
;
318 pa_log_debug("Write index deviates by %0.2f ms, expected %0.2f ms", (double) latency
/PA_USEC_PER_MSEC
, (double) s
->intended_latency
/PA_USEC_PER_MSEC
);
320 /* The buffer is filling with some unknown rate R̂ samples/second. If the rate of reading in
321 * the last T seconds was Rⁿ, then the increase in buffer latency ΔLⁿ = Lⁿ - Lⁿ⁻ⁱ in that
322 * same period is ΔLⁿ = (TR̂ - TRⁿ) / R̂, giving the estimated target rate
324 * R̂ = ─────────────── Rⁿ . (1)
327 * Setting the sample rate to R̂ results in the latency being constant (if the estimate of R̂
328 * is correct). But there is also the requirement to keep the buffer at a predefined target
329 * latency L̂. So instead of setting Rⁿ⁺ⁱ to R̂ immediately, the strategy will be to reduce R
330 * from Rⁿ⁺ⁱ to R̂ in a steps of T seconds, where Rⁿ⁺ⁱ is chosen such that in the total time
331 * aT the latency is reduced from Lⁿ to L̂. This strategy translates to the requirements
332 * ₐ R̂ - Rⁿ⁺ʲ a-j+1 j-1
333 * Σ T ────────── = L̂ - Lⁿ with Rⁿ⁺ʲ = ───── Rⁿ⁺ⁱ + ───── R̂ .
335 * Solving for Rⁿ⁺ⁱ gives
337 * Rⁿ⁺ⁱ = ───────────────── R̂ . (2)
339 * In the code below a = 7 is used.
341 * Equation (1) is not directly used in (2), but instead an exponentially weighted average
342 * of the estimated rate R̂ is used. This average R̅ is defined as
343 * R̅ⁿ = α R̂ⁿ + (1-α) R̅ⁿ⁻ⁱ .
344 * Because it is difficult to find a fixed value for the coefficient α such that the
345 * averaging is without significant lag but oscillations are filtered out, a heuristic is
346 * used. When the successive estimates R̂ⁿ do not change much then α→1, but when there is a
347 * sudden spike in the estimated rate α→0, such that the deviation is given little weight.
349 estimated_rate
= (double) current_rate
* (double) RATE_UPDATE_INTERVAL
/ (double) (RATE_UPDATE_INTERVAL
+ s
->last_latency
- latency
);
350 if (fabs(s
->estimated_rate
- s
->avg_estimated_rate
) > 1) {
351 double ratio
= (estimated_rate
+ s
->estimated_rate
- 2*s
->avg_estimated_rate
) / (s
->estimated_rate
- s
->avg_estimated_rate
);
352 alpha
= PA_CLAMP(2 * (ratio
+ fabs(ratio
)) / (4 + ratio
*ratio
), 0.02, 0.8);
354 s
->avg_estimated_rate
= alpha
* estimated_rate
+ (1-alpha
) * s
->avg_estimated_rate
;
355 s
->estimated_rate
= estimated_rate
;
356 pa_log_debug("Estimated target rate: %.0f Hz, using average of %.0f Hz (α=%.3f)", estimated_rate
, s
->avg_estimated_rate
, alpha
);
357 new_rate
= (uint32_t) ((double) (RATE_UPDATE_INTERVAL
+ latency
/4 - s
->intended_latency
/4) / (double) RATE_UPDATE_INTERVAL
* s
->avg_estimated_rate
);
358 s
->last_latency
= latency
;
360 if (new_rate
< (uint32_t) (base_rate
*0.8) || new_rate
> (uint32_t) (base_rate
*1.25)) {
361 pa_log_warn("Sample rates too different, not adjusting (%u vs. %u).", base_rate
, new_rate
);
362 new_rate
= base_rate
;
364 if (base_rate
< new_rate
+ 20 && new_rate
< base_rate
+ 20)
365 new_rate
= base_rate
;
366 /* Do the adjustment in small steps; 2‰ can be considered inaudible */
367 if (new_rate
< (uint32_t) (current_rate
*0.998) || new_rate
> (uint32_t) (current_rate
*1.002)) {
368 pa_log_info("New rate of %u Hz not within 2‰ of %u Hz, forcing smaller adjustment", new_rate
, current_rate
);
369 new_rate
= PA_CLAMP(new_rate
, (uint32_t) (current_rate
*0.998), (uint32_t) (current_rate
*1.002));
372 s
->sink_input
->sample_spec
.rate
= new_rate
;
374 pa_assert(pa_sample_spec_valid(&s
->sink_input
->sample_spec
));
376 pa_resampler_set_input_rate(s
->sink_input
->thread_info
.resampler
, s
->sink_input
->sample_spec
.rate
);
378 pa_log_debug("Updated sampling rate to %lu Hz.", (unsigned long) s
->sink_input
->sample_spec
.rate
);
380 s
->last_rate_update
= pa_timeval_load(&now
);
383 if (pa_memblockq_is_readable(s
->memblockq
) &&
384 s
->sink_input
->thread_info
.underrun_for
> 0) {
385 pa_log_debug("Requesting rewind due to end of underrun");
386 pa_sink_input_request_rewind(s
->sink_input
, 0, FALSE
, TRUE
, FALSE
);
392 /* Called from I/O thread context */
393 static void sink_input_attach(pa_sink_input
*i
) {
397 pa_sink_input_assert_ref(i
);
398 pa_assert_se(s
= i
->userdata
);
400 pa_assert(!s
->rtpoll_item
);
401 s
->rtpoll_item
= pa_rtpoll_item_new(i
->sink
->thread_info
.rtpoll
, PA_RTPOLL_LATE
, 1);
403 p
= pa_rtpoll_item_get_pollfd(s
->rtpoll_item
, NULL
);
404 p
->fd
= s
->rtp_context
.fd
;
408 pa_rtpoll_item_set_work_callback(s
->rtpoll_item
, rtpoll_work_cb
);
409 pa_rtpoll_item_set_userdata(s
->rtpoll_item
, s
);
412 /* Called from I/O thread context */
413 static void sink_input_detach(pa_sink_input
*i
) {
415 pa_sink_input_assert_ref(i
);
416 pa_assert_se(s
= i
->userdata
);
418 pa_assert(s
->rtpoll_item
);
419 pa_rtpoll_item_free(s
->rtpoll_item
);
420 s
->rtpoll_item
= NULL
;
423 static int mcast_socket(const struct sockaddr
* sa
, socklen_t salen
) {
424 int af
, fd
= -1, r
, one
;
427 pa_assert(salen
> 0);
430 if ((fd
= pa_socket_cloexec(af
, SOCK_DGRAM
, 0)) < 0) {
431 pa_log("Failed to create socket: %s", pa_cstrerror(errno
));
435 pa_make_udp_socket_low_delay(fd
);
438 if (setsockopt(fd
, SOL_SOCKET
, SO_TIMESTAMP
, &one
, sizeof(one
)) < 0) {
439 pa_log("SO_TIMESTAMP failed: %s", pa_cstrerror(errno
));
444 if (setsockopt(fd
, SOL_SOCKET
, SO_REUSEADDR
, &one
, sizeof(one
)) < 0) {
445 pa_log("SO_REUSEADDR failed: %s", pa_cstrerror(errno
));
451 memset(&mr4
, 0, sizeof(mr4
));
452 mr4
.imr_multiaddr
= ((const struct sockaddr_in
*) sa
)->sin_addr
;
453 r
= setsockopt(fd
, IPPROTO_IP
, IP_ADD_MEMBERSHIP
, &mr4
, sizeof(mr4
));
456 struct ipv6_mreq mr6
;
457 memset(&mr6
, 0, sizeof(mr6
));
458 mr6
.ipv6mr_multiaddr
= ((const struct sockaddr_in6
*) sa
)->sin6_addr
;
459 r
= setsockopt(fd
, IPPROTO_IPV6
, IPV6_JOIN_GROUP
, &mr6
, sizeof(mr6
));
464 pa_log_info("Joining mcast group failed: %s", pa_cstrerror(errno
));
468 if (bind(fd
, sa
, salen
) < 0) {
469 pa_log("bind() failed: %s", pa_cstrerror(errno
));
482 static struct session
*session_new(struct userdata
*u
, const pa_sdp_info
*sdp_info
) {
483 struct session
*s
= NULL
;
487 pa_sink_input_new_data data
;
493 if (u
->n_sessions
>= MAX_SESSIONS
) {
494 pa_log("Session limit reached.");
498 if (!(sink
= pa_namereg_get(u
->module
->core
, u
->sink_name
, PA_NAMEREG_SINK
))) {
499 pa_log("Sink does not exist.");
503 pa_rtclock_get(&now
);
505 s
= pa_xnew0(struct session
, 1);
507 s
->first_packet
= FALSE
;
508 s
->sdp_info
= *sdp_info
;
509 s
->rtpoll_item
= NULL
;
510 s
->intended_latency
= LATENCY_USEC
;
511 s
->smoother
= pa_smoother_new(
517 pa_timeval_load(&now
),
519 s
->last_rate_update
= pa_timeval_load(&now
);
520 s
->last_latency
= LATENCY_USEC
;
521 s
->estimated_rate
= (double) sink
->sample_spec
.rate
;
522 s
->avg_estimated_rate
= (double) sink
->sample_spec
.rate
;
523 pa_atomic_store(&s
->timestamp
, (int) now
.tv_sec
);
525 if ((fd
= mcast_socket((const struct sockaddr
*) &sdp_info
->sa
, sdp_info
->salen
)) < 0)
528 pa_sink_input_new_data_init(&data
);
530 data
.driver
= __FILE__
;
531 pa_proplist_sets(data
.proplist
, PA_PROP_MEDIA_ROLE
, "stream");
532 pa_proplist_setf(data
.proplist
, PA_PROP_MEDIA_NAME
,
534 sdp_info
->session_name
? " (" : "",
535 sdp_info
->session_name
? sdp_info
->session_name
: "",
536 sdp_info
->session_name
? ")" : "");
538 if (sdp_info
->session_name
)
539 pa_proplist_sets(data
.proplist
, "rtp.session", sdp_info
->session_name
);
540 pa_proplist_sets(data
.proplist
, "rtp.origin", sdp_info
->origin
);
541 pa_proplist_setf(data
.proplist
, "rtp.payload", "%u", (unsigned) sdp_info
->payload
);
542 data
.module
= u
->module
;
543 pa_sink_input_new_data_set_sample_spec(&data
, &sdp_info
->sample_spec
);
544 data
.flags
= PA_SINK_INPUT_VARIABLE_RATE
;
546 pa_sink_input_new(&s
->sink_input
, u
->module
->core
, &data
);
547 pa_sink_input_new_data_done(&data
);
549 if (!s
->sink_input
) {
550 pa_log("Failed to create sink input.");
554 s
->sink_input
->userdata
= s
;
556 s
->sink_input
->parent
.process_msg
= sink_input_process_msg
;
557 s
->sink_input
->pop
= sink_input_pop_cb
;
558 s
->sink_input
->process_rewind
= sink_input_process_rewind_cb
;
559 s
->sink_input
->update_max_rewind
= sink_input_update_max_rewind_cb
;
560 s
->sink_input
->kill
= sink_input_kill
;
561 s
->sink_input
->attach
= sink_input_attach
;
562 s
->sink_input
->detach
= sink_input_detach
;
563 s
->sink_input
->suspend_within_thread
= sink_input_suspend_within_thread
;
565 pa_sink_input_get_silence(s
->sink_input
, &silence
);
567 s
->sink_latency
= pa_sink_input_set_requested_latency(s
->sink_input
, s
->intended_latency
/2);
569 if (s
->intended_latency
< s
->sink_latency
*2)
570 s
->intended_latency
= s
->sink_latency
*2;
572 s
->memblockq
= pa_memblockq_new(
576 pa_frame_size(&s
->sink_input
->sample_spec
),
577 pa_usec_to_bytes(s
->intended_latency
- s
->sink_latency
, &s
->sink_input
->sample_spec
),
582 pa_memblock_unref(silence
.memblock
);
584 pa_rtp_context_init_recv(&s
->rtp_context
, fd
, pa_frame_size(&s
->sdp_info
.sample_spec
));
586 pa_hashmap_put(s
->userdata
->by_origin
, s
->sdp_info
.origin
, s
);
588 PA_LLIST_PREPEND(struct session
, s
->userdata
->sessions
, s
);
590 pa_sink_input_put(s
->sink_input
);
592 pa_log_info("New session '%s'", s
->sdp_info
.session_name
);
605 static void session_free(struct session
*s
) {
608 pa_log_info("Freeing session '%s'", s
->sdp_info
.session_name
);
610 pa_sink_input_unlink(s
->sink_input
);
611 pa_sink_input_unref(s
->sink_input
);
613 PA_LLIST_REMOVE(struct session
, s
->userdata
->sessions
, s
);
614 pa_assert(s
->userdata
->n_sessions
>= 1);
615 s
->userdata
->n_sessions
--;
616 pa_hashmap_remove(s
->userdata
->by_origin
, s
->sdp_info
.origin
);
618 pa_memblockq_free(s
->memblockq
);
619 pa_sdp_info_destroy(&s
->sdp_info
);
620 pa_rtp_context_destroy(&s
->rtp_context
);
622 pa_smoother_free(s
->smoother
);
627 static void sap_event_cb(pa_mainloop_api
*m
, pa_io_event
*e
, int fd
, pa_io_event_flags_t flags
, void *userdata
) {
628 struct userdata
*u
= userdata
;
629 pa_bool_t goodbye
= FALSE
;
636 pa_assert(fd
== u
->sap_context
.fd
);
637 pa_assert(flags
== PA_IO_EVENT_INPUT
);
639 if (pa_sap_recv(&u
->sap_context
, &goodbye
) < 0)
642 if (!pa_sdp_parse(u
->sap_context
.sdp_data
, &info
, goodbye
))
647 if ((s
= pa_hashmap_get(u
->by_origin
, info
.origin
)))
650 pa_sdp_info_destroy(&info
);
653 if (!(s
= pa_hashmap_get(u
->by_origin
, info
.origin
))) {
654 if (!session_new(u
, &info
))
655 pa_sdp_info_destroy(&info
);
659 pa_rtclock_get(&now
);
660 pa_atomic_store(&s
->timestamp
, (int) now
.tv_sec
);
662 pa_sdp_info_destroy(&info
);
667 static void check_death_event_cb(pa_mainloop_api
*m
, pa_time_event
*t
, const struct timeval
*tv
, void *userdata
) {
668 struct session
*s
, *n
;
669 struct userdata
*u
= userdata
;
676 pa_rtclock_get(&now
);
678 pa_log_debug("Checking for dead streams ...");
680 for (s
= u
->sessions
; s
; s
= n
) {
684 k
= pa_atomic_load(&s
->timestamp
);
686 if (k
+ DEATH_TIMEOUT
< now
.tv_sec
)
691 pa_core_rttime_restart(u
->module
->core
, t
, pa_rtclock_now() + DEATH_TIMEOUT
* PA_USEC_PER_SEC
);
694 int pa__init(pa_module
*m
) {
696 pa_modargs
*ma
= NULL
;
697 struct sockaddr_in sa4
;
699 struct sockaddr_in6 sa6
;
703 const char *sap_address
;
708 if (!(ma
= pa_modargs_new(m
->argument
, valid_modargs
))) {
709 pa_log("failed to parse module arguments");
713 sap_address
= pa_modargs_get_value(ma
, "sap_address", DEFAULT_SAP_ADDRESS
);
715 if (inet_pton(AF_INET
, sap_address
, &sa4
.sin_addr
) > 0) {
716 sa4
.sin_family
= AF_INET
;
717 sa4
.sin_port
= htons(SAP_PORT
);
718 sa
= (struct sockaddr
*) &sa4
;
721 } else if (inet_pton(AF_INET6
, sap_address
, &sa6
.sin6_addr
) > 0) {
722 sa6
.sin6_family
= AF_INET6
;
723 sa6
.sin6_port
= htons(SAP_PORT
);
724 sa
= (struct sockaddr
*) &sa6
;
728 pa_log("Invalid SAP address '%s'", sap_address
);
732 if ((fd
= mcast_socket(sa
, salen
)) < 0)
735 m
->userdata
= u
= pa_xnew(struct userdata
, 1);
738 u
->sink_name
= pa_xstrdup(pa_modargs_get_value(ma
, "sink", NULL
));
740 u
->sap_event
= m
->core
->mainloop
->io_new(m
->core
->mainloop
, fd
, PA_IO_EVENT_INPUT
, sap_event_cb
, u
);
741 pa_sap_context_init_recv(&u
->sap_context
, fd
);
743 PA_LLIST_HEAD_INIT(struct session
, u
->sessions
);
745 u
->by_origin
= pa_hashmap_new(pa_idxset_string_hash_func
, pa_idxset_string_compare_func
);
747 u
->check_death_event
= pa_core_rttime_new(m
->core
, pa_rtclock_now() + DEATH_TIMEOUT
* PA_USEC_PER_SEC
, check_death_event_cb
, u
);
763 void pa__done(pa_module
*m
) {
769 if (!(u
= m
->userdata
))
773 m
->core
->mainloop
->io_free(u
->sap_event
);
775 if (u
->check_death_event
)
776 m
->core
->mainloop
->time_free(u
->check_death_event
);
778 pa_sap_context_destroy(&u
->sap_context
);
781 while ((s
= pa_hashmap_first(u
->by_origin
)))
784 pa_hashmap_free(u
->by_origin
, NULL
, NULL
);
787 pa_xfree(u
->sink_name
);