diff options
author | Lennart Poettering <lennart@poettering.net> | 2008-05-09 22:48:37 +0000 |
---|---|---|
committer | Lennart Poettering <lennart@poettering.net> | 2008-05-09 22:48:37 +0000 |
commit | df92b23fa6e520127309c2f63e1f22c7d222e734 (patch) | |
tree | e12de600757d4ab8b76cef7828b2a9c604540af7 | |
parent | 580d56358d9d15792613fc4be886c71059c58a36 (diff) |
- Fix moving of sink inputs between sinks
- Don't write more than a single buffer size in the ALSA driver at a time, to give the clients time to fill up the memblockq again
- Add API for querying the requested latency of a sink input/source output
- Drop get_letancy() from vtable of sinks/sources
git-svn-id: file:///home/lennart/svn/public/pulseaudio/branches/glitch-free@2392 fefdeb5f-60dc-0310-8127-8f9354f1896f
-rw-r--r-- | src/modules/module-alsa-sink.c | 335 | ||||
-rw-r--r-- | src/modules/module-alsa-source.c | 2 | ||||
-rw-r--r-- | src/modules/module-combine.c | 44 | ||||
-rw-r--r-- | src/modules/module-esound-sink.c | 2 | ||||
-rw-r--r-- | src/modules/module-rescue-streams.c | 2 | ||||
-rw-r--r-- | src/modules/module-tunnel.c | 38 | ||||
-rw-r--r-- | src/pulse/stream.c | 34 | ||||
-rw-r--r-- | src/pulsecore/cli-command.c | 2 | ||||
-rw-r--r-- | src/pulsecore/cli-text.c | 24 | ||||
-rw-r--r-- | src/pulsecore/protocol-native.c | 67 | ||||
-rw-r--r-- | src/pulsecore/sink-input.c | 190 | ||||
-rw-r--r-- | src/pulsecore/sink-input.h | 23 | ||||
-rw-r--r-- | src/pulsecore/sink.c | 131 | ||||
-rw-r--r-- | src/pulsecore/sink.h | 8 | ||||
-rw-r--r-- | src/pulsecore/source-output.c | 41 | ||||
-rw-r--r-- | src/pulsecore/source-output.h | 25 | ||||
-rw-r--r-- | src/pulsecore/source.c | 4 | ||||
-rw-r--r-- | src/pulsecore/source.h | 1 |
18 files changed, 528 insertions, 445 deletions
diff --git a/src/modules/module-alsa-sink.c b/src/modules/module-alsa-sink.c index 4a997cd1..97e1e59f 100644 --- a/src/modules/module-alsa-sink.c +++ b/src/modules/module-alsa-sink.c @@ -131,6 +131,7 @@ struct userdata { pa_smoother *smoother; int64_t frame_index; + uint64_t since_start; snd_pcm_sframes_t hwbuf_unused_frames; }; @@ -162,6 +163,32 @@ static void fix_tsched_watermark(struct userdata *u) { u->tsched_watermark = min_wakeup; } +static void hw_sleep_time(struct userdata *u, pa_usec_t *sleep_usec, pa_usec_t*process_usec) { + pa_usec_t usec, wm; + + pa_assert(sleep_usec); + pa_assert(process_usec); + + pa_assert(u); + + usec = pa_sink_get_requested_latency_within_thread(u->sink); + + if (usec == (pa_usec_t) -1) + usec = pa_bytes_to_usec(u->hwbuf_size, &u->sink->sample_spec); + +/* pa_log_debug("hw buffer time: %u ms", (unsigned) (usec / PA_USEC_PER_MSEC)); */ + + wm = pa_bytes_to_usec(u->tsched_watermark, &u->sink->sample_spec); + + if (usec >= wm) { + *sleep_usec = usec - wm; + *process_usec = wm; + } else + *process_usec = *sleep_usec = usec / 2; + +/* pa_log_debug("after watermark: %u ms", (unsigned) (*sleep_usec / PA_USEC_PER_MSEC)); */ +} + static int try_recover(struct userdata *u, const char *call, int err) { pa_assert(u); pa_assert(call); @@ -169,16 +196,14 @@ static int try_recover(struct userdata *u, const char *call, int err) { pa_log_debug("%s: %s", call, snd_strerror(err)); - if (err == -EAGAIN) { - pa_log_debug("%s: EAGAIN", call); - return 1; - } + pa_assert(err != -EAGAIN); if (err == -EPIPE) pa_log_debug("%s: Buffer underrun!", call); if ((err = snd_pcm_recover(u->pcm_handle, err, 1)) == 0) { u->first = TRUE; + u->since_start = 0; return 0; } @@ -186,20 +211,17 @@ static int try_recover(struct userdata *u, const char *call, int err) { return -1; } -static void check_left_to_play(struct userdata *u, snd_pcm_sframes_t n) { +static size_t check_left_to_play(struct userdata *u, snd_pcm_sframes_t n) { size_t left_to_play; - if (u->first || u->after_rewind) - return; - if (n*u->frame_size < u->hwbuf_size) left_to_play = u->hwbuf_size - (n*u->frame_size); else left_to_play = 0; - if (left_to_play > 0) - pa_log_debug("%0.2f ms left to play", (double) pa_bytes_to_usec(left_to_play, &u->sink->sample_spec) / PA_USEC_PER_MSEC); - else { + if (left_to_play > 0) { +/* pa_log_debug("%0.2f ms left to play", (double) pa_bytes_to_usec(left_to_play, &u->sink->sample_spec) / PA_USEC_PER_MSEC); */ + } else if (!u->first && !u->after_rewind) { pa_log_info("Underrun!"); if (u->use_tsched) { @@ -213,22 +235,24 @@ static void check_left_to_play(struct userdata *u, snd_pcm_sframes_t n) { (double) pa_bytes_to_usec(u->tsched_watermark, &u->sink->sample_spec) / PA_USEC_PER_MSEC); } } + + return left_to_play; } -static int mmap_write(struct userdata *u) { +static int mmap_write(struct userdata *u, pa_usec_t *sleep_usec) { int work_done = 0; - pa_bool_t checked_left_to_play = FALSE; + pa_usec_t max_sleep_usec, process_usec; + size_t left_to_play; pa_assert(u); pa_sink_assert_ref(u->sink); + if (u->use_tsched) + hw_sleep_time(u, &max_sleep_usec, &process_usec); + for (;;) { - pa_memchunk chunk; - void *p; snd_pcm_sframes_t n; - int err, r; - const snd_pcm_channel_area_t *areas; - snd_pcm_uframes_t offset, frames; + int r; snd_pcm_hwsync(u->pcm_handle); @@ -239,92 +263,110 @@ static int mmap_write(struct userdata *u) { if ((r = try_recover(u, "snd_pcm_avail_update", n)) == 0) continue; - else if (r > 0) - return work_done; return r; } - if (!checked_left_to_play) { - check_left_to_play(u, n); - checked_left_to_play = TRUE; - } + left_to_play = check_left_to_play(u, n); + + if (u->use_tsched) + + /* We won't fill up the playback buffer before at least + * half the sleep time is over because otherwise we might + * ask for more data from the clients then they expect. We + * need to guarantee that clients only have to keep around + * a single hw buffer length. */ - /* We only use part of the buffer that matches our - * dynamically requested latency */ + if (pa_bytes_to_usec(left_to_play, &u->sink->sample_spec) > max_sleep_usec/2) + break; if (PA_UNLIKELY(n <= u->hwbuf_unused_frames)) - return work_done; + break; - frames = n = n - u->hwbuf_unused_frames; + n -= u->hwbuf_unused_frames; -/* pa_log_debug("%lu frames to write", (unsigned long) frames);*/ +/* pa_log_debug("Filling up"); */ - if (PA_UNLIKELY((err = snd_pcm_mmap_begin(u->pcm_handle, &areas, &offset, &frames)) < 0)) { + for (;;) { + pa_memchunk chunk; + void *p; + int err; + const snd_pcm_channel_area_t *areas; + snd_pcm_uframes_t offset, frames = (snd_pcm_uframes_t) n; - if ((r = try_recover(u, "snd_pcm_mmap_begin", err)) == 0) - continue; - else if (r > 0) - return work_done; +/* pa_log_debug("%lu frames to write", (unsigned long) frames); */ - return r; - } + if (PA_UNLIKELY((err = snd_pcm_mmap_begin(u->pcm_handle, &areas, &offset, &frames)) < 0)) { - /* Make sure that if these memblocks need to be copied they will fit into one slot */ - if (frames > pa_mempool_block_size_max(u->sink->core->mempool)/u->frame_size) - frames = pa_mempool_block_size_max(u->sink->core->mempool)/u->frame_size; + if ((r = try_recover(u, "snd_pcm_mmap_begin", err)) == 0) + continue; - /* Check these are multiples of 8 bit */ - pa_assert((areas[0].first & 7) == 0); - pa_assert((areas[0].step & 7)== 0); + return r; + } - /* We assume a single interleaved memory buffer */ - pa_assert((areas[0].first >> 3) == 0); - pa_assert((areas[0].step >> 3) == u->frame_size); + /* Make sure that if these memblocks need to be copied they will fit into one slot */ + if (frames > pa_mempool_block_size_max(u->sink->core->mempool)/u->frame_size) + frames = pa_mempool_block_size_max(u->sink->core->mempool)/u->frame_size; - p = (uint8_t*) areas[0].addr + (offset * u->frame_size); + /* Check these are multiples of 8 bit */ + pa_assert((areas[0].first & 7) == 0); + pa_assert((areas[0].step & 7)== 0); - chunk.memblock = pa_memblock_new_fixed(u->core->mempool, p, frames * u->frame_size, TRUE); - chunk.length = pa_memblock_get_length(chunk.memblock); - chunk.index = 0; + /* We assume a single interleaved memory buffer */ + pa_assert((areas[0].first >> 3) == 0); + pa_assert((areas[0].step >> 3) == u->frame_size); - pa_sink_render_into_full(u->sink, &chunk); + p = (uint8_t*) areas[0].addr + (offset * u->frame_size); - /* FIXME: Maybe we can do something to keep this memory block - * a little bit longer around? */ - pa_memblock_unref_fixed(chunk.memblock); + chunk.memblock = pa_memblock_new_fixed(u->core->mempool, p, frames * u->frame_size, TRUE); + chunk.length = pa_memblock_get_length(chunk.memblock); + chunk.index = 0; - if (PA_UNLIKELY((err = snd_pcm_mmap_commit(u->pcm_handle, offset, frames)) < 0)) { + pa_sink_render_into_full(u->sink, &chunk); - if ((r = try_recover(u, "snd_pcm_mmap_commit", err)) == 0) - continue; - else if (r > 0) - return work_done; + /* FIXME: Maybe we can do something to keep this memory block + * a little bit longer around? */ + pa_memblock_unref_fixed(chunk.memblock); - return r; - } + if (PA_UNLIKELY((err = snd_pcm_mmap_commit(u->pcm_handle, offset, frames)) < 0)) { - work_done = 1; + if ((r = try_recover(u, "snd_pcm_mmap_commit", err)) == 0) + continue; - u->frame_index += frames; + return r; + } -/* pa_log_debug("wrote %lu frames", (unsigned long) frames); */ + work_done = 1; + + u->frame_index += frames; + u->since_start += frames * u->frame_size; + +/* pa_log_debug("wrote %lu frames", (unsigned long) frames); */ - if (PA_LIKELY(frames >= (snd_pcm_uframes_t) n)) - return work_done; + if (frames >= (snd_pcm_uframes_t) n) + break; + + n -= frames; + } } + + *sleep_usec = pa_bytes_to_usec(left_to_play, &u->sink->sample_spec) - process_usec; + return work_done; } -static int unix_write(struct userdata *u) { +static int unix_write(struct userdata *u, pa_usec_t *sleep_usec) { int work_done = 0; - pa_bool_t checked_left_to_play = FALSE; + pa_usec_t max_sleep_usec, process_usec; + size_t left_to_play; pa_assert(u); pa_sink_assert_ref(u->sink); + if (u->use_tsched) + hw_sleep_time(u, &max_sleep_usec, &process_usec); + for (;;) { - void *p; - snd_pcm_sframes_t n, frames; + snd_pcm_sframes_t n; int r; snd_pcm_hwsync(u->pcm_handle); @@ -333,67 +375,82 @@ static int unix_write(struct userdata *u) { if ((r = try_recover(u, "snd_pcm_avail_update", n)) == 0) continue; - else if (r > 0) - return work_done; return r; } - if (!checked_left_to_play) { - check_left_to_play(u, n); - checked_left_to_play = TRUE; - } + left_to_play = check_left_to_play(u, n); + + if (u->use_tsched) + + /* We won't fill up the playback buffer before at least + * half the sleep time is over because otherwise we might + * ask for more data from the clients then they expect. We + * need to guarantee that clients only have to keep around + * a single hw buffer length. */ + + if (pa_bytes_to_usec(left_to_play, &u->sink->sample_spec) > max_sleep_usec/2) + break; if (PA_UNLIKELY(n <= u->hwbuf_unused_frames)) - return work_done; + break; n -= u->hwbuf_unused_frames; + for (;;) { + snd_pcm_sframes_t frames; + void *p; + /* pa_log_debug("%lu frames to write", (unsigned long) frames); */ - if (u->memchunk.length <= 0) - pa_sink_render(u->sink, n * u->frame_size, &u->memchunk); + if (u->memchunk.length <= 0) + pa_sink_render(u->sink, n * u->frame_size, &u->memchunk); - pa_assert(u->memchunk.length > 0); + pa_assert(u->memchunk.length > 0); - frames = u->memchunk.length / u->frame_size; + frames = u->memchunk.length / u->frame_size; - if (frames > n) - frames = n; + if (frames > n) + frames = n; - p = pa_memblock_acquire(u->memchunk.memblock); - frames = snd_pcm_writei(u->pcm_handle, (const uint8_t*) p + u->memchunk.index, frames); - pa_memblock_release(u->memchunk.memblock); + p = pa_memblock_acquire(u->memchunk.memblock); + frames = snd_pcm_writei(u->pcm_handle, (const uint8_t*) p + u->memchunk.index, frames); + pa_memblock_release(u->memchunk.memblock); - pa_assert(frames != 0); + pa_assert(frames != 0); - if (PA_UNLIKELY(frames < 0)) { + if (PA_UNLIKELY(frames < 0)) { - if ((r = try_recover(u, "snd_pcm_writei", n)) == 0) - continue; - else if (r > 0) - return work_done; + if ((r = try_recover(u, "snd_pcm_writei", n)) == 0) + continue; - return r; - } + return r; + } - u->memchunk.index += frames * u->frame_size; - u->memchunk.length -= frames * u->frame_size; + u->memchunk.index += frames * u->frame_size; + u->memchunk.length -= frames * u->frame_size; - if (u->memchunk.length <= 0) { - pa_memblock_unref(u->memchunk.memblock); - pa_memchunk_reset(&u->memchunk); - } + if (u->memchunk.length <= 0) { + pa_memblock_unref(u->memchunk.memblock); + pa_memchunk_reset(&u->memchunk); + } - work_done = 1; + work_done = 1; - u->frame_index += frames; + u->frame_index += frames; + u->since_start += frames * u->frame_size; /* pa_log_debug("wrote %lu frames", (unsigned long) frames); */ - if (PA_LIKELY(frames >= n)) - return work_done; + if (frames >= n) + break; + + n -= frames; + } } + + *sleep_usec = pa_bytes_to_usec(left_to_play, &u->sink->sample_spec) - process_usec; + return work_done; } static void update_smoother(struct userdata *u) { @@ -494,35 +551,6 @@ static int suspend(struct userdata *u) { return 0; } -static pa_usec_t hw_sleep_time(struct userdata *u) { - pa_usec_t usec, wm; - - pa_assert(u); - - usec = pa_sink_get_requested_latency_within_thread(u->sink); - - if (usec == (pa_usec_t) -1) - usec = pa_bytes_to_usec(u->hwbuf_size, &u->sink->sample_spec); - - pa_log_debug("hw buffer time: %u ms", (unsigned) (usec / PA_USEC_PER_MSEC)); - - wm = pa_bytes_to_usec(u->tsched_watermark, &u->sink->sample_spec); - - if (usec >= wm) - usec -= wm; - else - usec /= 2; - - if (u->first) { - pa_log_debug("Decreasing wakeup time for the first iteration by half."); - usec /= 2; - } - - pa_log_debug("after watermark: %u ms", (unsigned) (usec / PA_USEC_PER_MSEC)); - - return usec; -} - static int update_sw_params(struct userdata *u) { snd_pcm_uframes_t avail_min; int err; @@ -561,10 +589,10 @@ static int update_sw_params(struct userdata *u) { avail_min = u->hwbuf_unused_frames + 1; if (u->use_tsched) { - pa_usec_t usec; + pa_usec_t sleep_usec, process_usec; - usec = hw_sleep_time(u); - avail_min += pa_usec_to_bytes(usec, &u->sink->sample_spec); + hw_sleep_time(u, &sleep_usec, &process_usec); + avail_min += pa_usec_to_bytes(sleep_usec, &u->sink->sample_spec); } pa_log_debug("setting avail_min=%lu", (unsigned long) avail_min); @@ -630,6 +658,7 @@ static int unsuspend(struct userdata *u) { /* FIXME: We need to reload the volume somehow */ u->first = TRUE; + u->since_start = 0; pa_log_info("Resumed successfully..."); @@ -932,16 +961,17 @@ static void thread_func(void *userdata) { /* Render some data and write it to the dsp */ if (PA_SINK_IS_OPENED(u->sink->thread_info.state)) { - int work_done = 0; + int work_done; + pa_usec_t sleep_usec; if (u->sink->thread_info.rewind_nbytes > 0) if (process_rewind(u) < 0) goto fail; if (u->use_mmap) - work_done = mmap_write(u); + work_done = mmap_write(u, &sleep_usec); else - work_done = unix_write(u); + work_done = unix_write(u, &sleep_usec); if (work_done < 0) goto fail; @@ -961,23 +991,34 @@ static void thread_func(void *userdata) { } if (u->use_tsched) { - pa_usec_t usec, cusec; + pa_usec_t cusec; - /* OK, the playback buffer is now full, let's - * calculate when to wake up next */ + if (u->since_start <= u->hwbuf_size) { + + /* USB devices on ALSA seem to hit a buffer + * underrun during the first iterations much + * quicker then we calculate here, probably due to + * the transport latency. To accomodate for that + * we artificially decrease the sleep time until + * we have filled the buffer at least once + * completely.*/ - usec = hw_sleep_time(u); + pa_log_debug("Cutting sleep time for the initial iterations by half."); + sleep_usec /= 2; + } -/* pa_log_debug("Waking up in %0.2fms (sound card clock).", (double) usec / PA_USEC_PER_MSEC); */ + /* OK, the playback buffer is now full, let's + * calculate when to wake up next */ +/* pa_log_debug("Waking up in %0.2fms (sound card clock).", (double) sleep_usec / PA_USEC_PER_MSEC); */ /* Convert from the sound card time domain to the * system time domain */ - cusec = pa_smoother_translate(u->smoother, pa_rtclock_usec(), usec); + cusec = pa_smoother_translate(u->smoother, pa_rtclock_usec(), sleep_usec); /* pa_log_debug("Waking up in %0.2fms (system clock).", (double) cusec / PA_USEC_PER_MSEC); */ /* We don't trust the conversion, so we wake up whatever comes first */ - pa_rtpoll_set_timer_relative(u->rtpoll, PA_MIN(usec, cusec)); + pa_rtpoll_set_timer_relative(u->rtpoll, PA_MIN(sleep_usec, cusec)); } u->first = FALSE; @@ -1014,6 +1055,7 @@ static void thread_func(void *userdata) { goto fail; u->first = TRUE; + u->since_start = 0; } if (revents) @@ -1115,12 +1157,13 @@ int pa__init(pa_module*m) { u->use_mmap = use_mmap; u->use_tsched = use_tsched; u->first = TRUE; + u->since_start = 0; u->after_rewind = FALSE; u->rtpoll = pa_rtpoll_new(); pa_thread_mq_init(&u->thread_mq, m->core->mainloop, u->rtpoll); u->alsa_rtpoll_item = NULL; - u->smoother = pa_smoother_new(DEFAULT_TSCHED_BUFFER_USEC*2, DEFAULT_TSCHED_BUFFER_USEC*2, TRUE); + u->smoother = pa_smoother_new(DEFAULT_TSCHED_BUFFER_USEC*2, DEFAULT_TSCHED_BUFFER_USEC*2, TRUE, 5); usec = pa_rtclock_usec(); pa_smoother_set_time_offset(u->smoother, usec); pa_smoother_pause(u->smoother, usec); diff --git a/src/modules/module-alsa-source.c b/src/modules/module-alsa-source.c index 9eb6f06b..4838ad27 100644 --- a/src/modules/module-alsa-source.c +++ b/src/modules/module-alsa-source.c @@ -988,7 +988,7 @@ int pa__init(pa_module*m) { pa_thread_mq_init(&u->thread_mq, m->core->mainloop, u->rtpoll); u->alsa_rtpoll_item = NULL; - u->smoother = pa_smoother_new(DEFAULT_TSCHED_WATERMARK_USEC, DEFAULT_TSCHED_WATERMARK_USEC, TRUE); + u->smoother = pa_smoother_new(DEFAULT_TSCHED_WATERMARK_USEC, DEFAULT_TSCHED_WATERMARK_USEC, TRUE, 5); pa_smoother_set_time_offset(u->smoother, pa_rtclock_usec()); snd_config_update_free_global(); diff --git a/src/modules/module-combine.c b/src/modules/module-combine.c index 2409ef8f..fc8be18d 100644 --- a/src/modules/module-combine.c +++ b/src/modules/module-combine.c @@ -618,35 +618,35 @@ static int sink_process_msg(pa_msgobject *o, int code, void *data, int64_t offse } /* Called from main context */ -static pa_usec_t sink_get_latency_cb(pa_sink *s) { - struct userdata *u; +/* static pa_usec_t sink_get_latency_cb(pa_sink *s) { */ +/* struct userdata *u; */ - pa_sink_assert_ref(s); - pa_assert_se(u = s->userdata); +/* pa_sink_assert_ref(s); */ +/* pa_assert_se(u = s->userdata); */ - if (u->master) { - /* If we have a master sink, we just return the latency of it - * and add our own buffering on top */ +/* if (u->master) { */ +/* /\* If we have a master sink, we just return the latency of it */ +/* * and add our own buffering on top *\/ */ - if (!u->master->sink_input) - return 0; +/* if (!u->master->sink_input) */ +/* return 0; */ - return - pa_sink_input_get_latency(u->master->sink_input) + - pa_sink_get_latency(u->master->sink); +/* return */ +/* pa_sink_input_get_latency(u->master->sink_input) + */ +/* pa_sink_get_latency(u->master->sink); */ - } else { - pa_usec_t usec = 0; +/* } else { */ +/* pa_usec_t usec = 0; */ - /* We have no master, hence let's ask our own thread which - * implements the NULL sink */ +/* /\* We have no master, hence let's ask our own thread which */ +/* * implements the NULL sink *\/ */ - if (pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SINK_MESSAGE_GET_LATENCY, &usec, 0, NULL) < 0) - return 0; +/* if (pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SINK_MESSAGE_GET_LATENCY, &usec, 0, NULL) < 0) */ +/* return 0; */ - return usec; - } -} +/* return usec; */ +/* } */ +/* } */ static void update_description(struct userdata *u) { int first = 1; @@ -1025,7 +1025,7 @@ int pa__init(pa_module*m) { } u->sink->parent.process_msg = sink_process_msg; - u->sink->get_latency = sink_get_latency_cb; +/* u->sink->get_latency = sink_get_latency_cb; */ u->sink->set_state = sink_set_state; u->sink->userdata = u; diff --git a/src/modules/module-esound-sink.c b/src/modules/module-esound-sink.c index 2206e2bc..4aa7d674 100644 --- a/src/modules/module-esound-sink.c +++ b/src/modules/module-esound-sink.c @@ -534,7 +534,7 @@ int pa__init(pa_module*m) { u->module = m; m->userdata = u; u->fd = -1; - u->smoother = pa_smoother_new(PA_USEC_PER_SEC, PA_USEC_PER_SEC*2, TRUE); + u->smoother = pa_smoother_new(PA_USEC_PER_SEC, PA_USEC_PER_SEC*2, TRUE, 10); pa_memchunk_reset(&u->memchunk); u->offset = 0; diff --git a/src/modules/module-rescue-streams.c b/src/modules/module-rescue-streams.c index dda54735..7241a99f 100644 --- a/src/modules/module-rescue-streams.c +++ b/src/modules/module-rescue-streams.c @@ -75,7 +75,7 @@ static pa_hook_result_t sink_hook_callback(pa_core *c, pa_sink *sink, void* user } while ((i = pa_idxset_first(sink->inputs, NULL))) { - if (pa_sink_input_move_to(i, target, 1) < 0) { + if (pa_sink_input_move_to(i, target) < 0) { pa_log_warn("Failed to move sink input %u \"%s\" to %s.", i->index, pa_proplist_gets(i->proplist, PA_PROP_APPLICATION_NAME), target->name); return PA_HOOK_OK; } diff --git a/src/modules/module-tunnel.c b/src/modules/module-tunnel.c index e3ae5e1f..7a87fd8c 100644 --- a/src/modules/module-tunnel.c +++ b/src/modules/module-tunnel.c @@ -577,29 +577,29 @@ static void timeout_callback(pa_mainloop_api *m, pa_time_event*e, PA_GCC_UNUSED } #ifdef TUNNEL_SINK -static pa_usec_t sink_get_latency(pa_sink *s) { - pa_usec_t t, c; - struct userdata *u = s->userdata; +/* static pa_usec_t sink_get_latency(pa_sink *s) { */ +/* pa_usec_t t, c; */ +/* struct userdata *u = s->userdata; */ - pa_sink_assert_ref(s); +/* pa_sink_assert_ref(s); */ - c = pa_bytes_to_usec(u->counter, &s->sample_spec); - t = pa_smoother_get(u->smoother, pa_rtclock_usec()); +/* c = pa_bytes_to_usec(u->counter, &s->sample_spec); */ +/* t = pa_smoother_get(u->smoother, pa_rtclock_usec()); */ - return c > t ? c - t : 0; -} +/* return c > t ? c - t : 0; */ +/* } */ #else -static pa_usec_t source_get_latency(pa_source *s) { - pa_usec_t t, c; - struct userdata *u = s->userdata; +/* static pa_usec_t source_get_latency(pa_source *s) { */ +/* pa_usec_t t, c; */ +/* struct userdata *u = s->userdata; */ - pa_source_assert_ref(s); +/* pa_source_assert_ref(s); */ - c = pa_bytes_to_usec(u->counter, &s->sample_spec); - t = pa_smoother_get(u->smoother, pa_rtclock_usec()); +/* c = pa_bytes_to_usec(u->counter, &s->sample_spec); */ +/* t = pa_smoother_get(u->smoother, pa_rtclock_usec()); */ - return t > c ? t - c : 0; -} +/* return t > c ? t - c : 0; */ +/* } */ #endif static void update_description(struct userdata *u) { @@ -1323,7 +1323,7 @@ int pa__init(pa_module*m) { u->source_name = pa_xstrdup(pa_modargs_get_value(ma, "source", NULL));; u->source = NULL; #endif - u->smoother = pa_smoother_new(PA_USEC_PER_SEC, PA_USEC_PER_SEC*2, TRUE); + u->smoother = pa_smoother_new(PA_USEC_PER_SEC, PA_USEC_PER_SEC*2, TRUE, 10); u->ctag = 1; u->device_index = u->channel = PA_INVALID_INDEX; u->auth_cookie_in_property = FALSE; @@ -1377,7 +1377,7 @@ int pa__init(pa_module*m) { u->sink->parent.process_msg = sink_process_msg; u->sink->userdata = u; u->sink->set_state = sink_set_state; - u->sink->get_latency = sink_get_latency; +/* u->sink->get_latency = sink_get_latency; */ u->sink->get_volume = sink_get_volume; u->sink->get_mute = sink_get_mute; u->sink->set_volume = sink_set_volume; @@ -1412,7 +1412,7 @@ int pa__init(pa_module*m) { u->source->parent.process_msg = source_process_msg; u->source->userdata = u; u->source->set_state = source_set_state; - u->source->get_latency = source_get_latency; +/* u->source->get_latency = source_get_latency; */ pa_source_set_asyncmsgq(u->source, u->thread_mq.inq); pa_source_set_rtpoll(u->source, u->rtpoll); diff --git a/src/pulse/stream.c b/src/pulse/stream.c index 3b1975fa..bd633cf0 100644 --- a/src/pulse/stream.c +++ b/src/pulse/stream.c @@ -42,10 +42,11 @@ #include "internal.h" -#define LATENCY_IPOL_INTERVAL_USEC (500*PA_USEC_PER_MSEC) +#define LATENCY_IPOL_INTERVAL_USEC (333*PA_USEC_PER_MSEC) #define SMOOTHER_ADJUST_TIME (1000*PA_USEC_PER_MSEC) #define SMOOTHER_HISTORY_TIME (5000*PA_USEC_PER_MSEC) +#define SMOOTHER_MIN_HISTORY (4) pa_stream *pa_stream_new(pa_context *c, const char *name, const pa_sample_spec *ss, const pa_channel_map *map) { return pa_stream_new_with_proplist(c, name, ss, map, NULL); @@ -344,6 +345,7 @@ void pa_command_stream_moved(pa_pdispatch *pd, uint32_t command, PA_GCC_UNUSED u pa_bool_t suspended; uint32_t di; pa_usec_t usec; + uint32_t maxlength = 0, fragsize = 0, minreq = 0, tlength = 0, prebuf = 0; pa_assert(pd); pa_assert(command == PA_COMMAND_PLAYBACK_STREAM_MOVED || command == PA_COMMAND_RECORD_STREAM_MOVED); @@ -367,14 +369,28 @@ void pa_command_stream_moved(pa_pdispatch *pd, uint32_t command, PA_GCC_UNUSED u } if (c->version >= 13) { - if (pa_tagstruct_get_usec(t, &usec) < 0) { - pa_context_fail(s->context, PA_ERR_PROTOCOL); - goto finish; + + if (s->direction == PA_STREAM_RECORD) { + if (pa_tagstruct_getu32(t, &maxlength) < 0 || + pa_tagstruct_getu32(t, &fragsize) < 0 || + pa_tagstruct_get_usec(t, &usec) < 0) { + pa_context_fail(c, PA_ERR_PROTOCOL); + goto finish; + } + } else { + if (pa_tagstruct_getu32(t, &maxlength) < 0 || + pa_tagstruct_getu32(t, &tlength) < 0 || + pa_tagstruct_getu32(t, &prebuf) < 0 || + pa_tagstruct_getu32(t, &minreq) < 0 || + pa_tagstruct_get_usec(t, &usec) < 0) { + pa_context_fail(c, PA_ERR_PROTOCOL); + goto finish; + } } } if (!pa_tagstruct_eof(t)) { - pa_context_fail(s->context, PA_ERR_PROTOCOL); + pa_context_fail(c, PA_ERR_PROTOCOL); goto finish; } @@ -394,6 +410,12 @@ void pa_command_stream_moved(pa_pdispatch *pd, uint32_t command, PA_GCC_UNUSED u s->timing_info.configured_source_usec = usec; else s->timing_info.configured_sink_usec = usec; + + s->buffer_attr.maxlength = maxlength; + s->buffer_attr.fragsize = fragsize; + s->buffer_attr.tlength = tlength; + s->buffer_attr.prebuf = prebuf; + s->buffer_attr.minreq = minreq; } pa_xfree(s->device_name); @@ -861,7 +883,7 @@ static int create_stream( if (s->smoother) pa_smoother_free(s->smoother); - s->smoother = pa_smoother_new(SMOOTHER_ADJUST_TIME, SMOOTHER_HISTORY_TIME, !(flags & PA_STREAM_NOT_MONOTONOUS)); + s->smoother = pa_smoother_new(SMOOTHER_ADJUST_TIME, SMOOTHER_HISTORY_TIME, !(flags & PA_STREAM_NOT_MONOTONOUS), SMOOTHER_MIN_HISTORY); x = pa_rtclock_usec(); pa_smoother_set_time_offset(s->smoother, x); diff --git a/src/pulsecore/cli-command.c b/src/pulsecore/cli-command.c index b6194f84..925a2e8c 100644 --- a/src/pulsecore/cli-command.c +++ b/src/pulsecore/cli-command.c @@ -1055,7 +1055,7 @@ static int pa_cli_command_move_sink_input(pa_core *c, pa_tokenizer *t, pa_strbuf return -1; } - if (pa_sink_input_move_to(si, sink, 0) < 0) { + if (pa_sink_input_move_to(si, sink) < 0) { pa_strbuf_puts(buf, "Moved failed.\n"); return -1; } diff --git a/src/pulsecore/cli-text.c b/src/pulsecore/cli-text.c index f251a146..029a7089 100644 --- a/src/pulsecore/cli-text.c +++ b/src/pulsecore/cli-text.c @@ -253,7 +253,13 @@ char *pa_source_output_list_to_string(pa_core *c) { pa_strbuf_printf(s, "%u source outputs(s) available.\n", pa_idxset_size(c->source_outputs)); for (o = pa_idxset_first(c->source_outputs, &idx); o; o = pa_idxset_next(c->source_outputs, &idx)) { - char ss[PA_SAMPLE_SPEC_SNPRINT_MAX], cm[PA_CHANNEL_MAP_SNPRINT_MAX], *t; + char ss[PA_SAMPLE_SPEC_SNPRINT_MAX], cm[PA_CHANNEL_MAP_SNPRINT_MAX], *t, clt[28]; + pa_usec_t cl; + + if ((cl = pa_source_output_get_requested_latency(o)) == (pa_usec_t) -1) + pa_snprintf(clt, sizeof(clt), "n/a"); + else + pa_snprintf(clt, sizeof(clt), "%0.2f ms", (double) cl / PA_USEC_PER_MSEC); pa_assert(o->source); @@ -264,7 +270,8 @@ char *pa_source_output_list_to_string(pa_core *c) { "\tflags: %s%s%s%s%s%s%s%s\n" "\tstate: %s\n" "\tsource: %u <%s>\n" - "\tlatency: %0.2f ms\n" + "\tcurrent latency: %0.2f ms\n" + "\trequested latency: %s\n" "\tsample spec: %s\n" "\tchannel map: %s\n" "\tresample method: %s\n", @@ -281,6 +288,7 @@ char *pa_source_output_list_to_string(pa_core *c) { state_table[pa_source_output_get_state(o)], o->source->index, o->source->name, (double) pa_source_output_get_latency(o) / PA_USEC_PER_MSEC, + clt, pa_sample_spec_snprint(ss, sizeof(ss), &o->sample_spec), pa_channel_map_snprint(cm, sizeof(cm), &o->channel_map), pa_resample_method_to_string(pa_source_output_get_resample_method(o))); @@ -315,7 +323,13 @@ char *pa_sink_input_list_to_string(pa_core *c) { pa_strbuf_printf(s, "%u sink input(s) available.\n", pa_idxset_size(c->sink_inputs)); for (i = pa_idxset_first(c->sink_inputs, &idx); i; i = pa_idxset_next(c->sink_inputs, &idx)) { - char ss[PA_SAMPLE_SPEC_SNPRINT_MAX], cv[PA_CVOLUME_SNPRINT_MAX], cm[PA_CHANNEL_MAP_SNPRINT_MAX], *t; + char ss[PA_SAMPLE_SPEC_SNPRINT_MAX], cv[PA_CVOLUME_SNPRINT_MAX], cm[PA_CHANNEL_MAP_SNPRINT_MAX], *t, clt[28]; + pa_usec_t cl; + + if ((cl = pa_sink_input_get_requested_latency(i)) == (pa_usec_t) -1) + pa_snprintf(clt, sizeof(clt), "n/a"); + else + pa_snprintf(clt, sizeof(clt), "%0.2f ms", (double) cl / PA_USEC_PER_MSEC); pa_assert(i->sink); @@ -328,7 +342,8 @@ char *pa_sink_input_list_to_string(pa_core *c) { "\tsink: %u <%s>\n" "\tvolume: %s\n" "\tmuted: %s\n" - "\tlatency: %0.2f ms\n" + "\tcurrent latency: %0.2f ms\n" + "\trequested latency: %s\n" "\tsample spec: %s\n" "\tchannel map: %s\n" "\tresample method: %s\n", @@ -347,6 +362,7 @@ char *pa_sink_input_list_to_string(pa_core *c) { pa_cvolume_snprint(cv, sizeof(cv), pa_sink_input_get_volume(i)), pa_yes_no(pa_sink_input_get_mute(i)), (double) pa_sink_input_get_latency(i) / PA_USEC_PER_MSEC, + clt, pa_sample_spec_snprint(ss, sizeof(ss), &i->sample_spec), pa_channel_map_snprint(cm, sizeof(cm), &i->channel_map), pa_resample_method_to_string(pa_sink_input_get_resample_method(i))); diff --git a/src/pulsecore/protocol-native.c b/src/pulsecore/protocol-native.c index 7eeefb84..a824b4da 100644 --- a/src/pulsecore/protocol-native.c +++ b/src/pulsecore/protocol-native.c @@ -818,6 +818,8 @@ static void fix_playback_buffer_attr_post(playback_stream *s, uint32_t *maxlengt *tlength = (uint32_t) pa_memblockq_get_tlength(s->memblockq); *prebuf = (uint32_t) pa_memblockq_get_prebuf(s->memblockq); *minreq = (uint32_t) pa_memblockq_get_minreq(s->memblockq); + + s->minreq = *minreq; } static playback_stream* playback_stream_new( @@ -933,7 +935,6 @@ static playback_stream* playback_stream_new( *ss = s->sink_input->sample_spec; *map = s->sink_input->channel_map; - s->minreq = *minreq; pa_atomic_store(&s->missing, 0); s->drain_request = FALSE; @@ -1290,14 +1291,14 @@ static int sink_input_pop_cb(pa_sink_input *i, size_t nbytes, pa_memchunk *chunk } else if (i->thread_info.playing_for > 0) pa_asyncmsgq_post(pa_thread_mq_get()->outq, PA_MSGOBJECT(s), PLAYBACK_STREAM_MESSAGE_UNDERFLOW, NULL, 0, NULL, NULL); -/* pa_log("adding %llu bytes, total is %llu", (unsigned long long) nbytes, (unsigned long long) i->thread_info.underrun_for); */ +/* pa_log("adding %llu bytes", (unsigned long long) nbytes); */ request_bytes(s); return -1; } -/* pa_log("NOTUNDERRUN"); */ +/* pa_log("NOTUNDERRUN %lu", (unsigned long) chunk->length); */ if (i->thread_info.underrun_for > 0) pa_asyncmsgq_post(pa_thread_mq_get()->outq, PA_MSGOBJECT(s), PLAYBACK_STREAM_MESSAGE_STARTED, NULL, 0, NULL, NULL); @@ -1368,11 +1369,24 @@ static void sink_input_suspend_cb(pa_sink_input *i, pa_bool_t suspend) { static void sink_input_moved_cb(pa_sink_input *i) { playback_stream *s; pa_tagstruct *t; + uint32_t maxlength, tlength, prebuf, minreq; pa_sink_input_assert_ref(i); s = PLAYBACK_STREAM(i->userdata); playback_stream_assert_ref(s); + maxlength = (uint32_t) pa_memblockq_get_maxlength(s->memblockq); + tlength = (uint32_t) pa_memblockq_get_tlength(s->memblockq); + prebuf = (uint32_t) pa_memblockq_get_prebuf(s->memblockq); + minreq = (uint32_t) pa_memblockq_get_minreq(s->memblockq); + + fix_playback_buffer_attr_pre(s, TRUE, &maxlength, &tlength, &prebuf, &minreq); + pa_memblockq_set_maxlength(s->memblockq, maxlength); + pa_memblockq_set_tlength(s->memblockq, tlength); + pa_memblockq_set_prebuf(s->memblockq, prebuf); + pa_memblockq_set_minreq(s->memblockq, minreq); + fix_playback_buffer_attr_post(s, &maxlength, &tlength, &prebuf, &minreq); + if (s->connection->version < 12) return; @@ -1383,6 +1397,15 @@ static void sink_input_moved_cb(pa_sink_input *i) { pa_tagstruct_putu32(t, i->sink->index); pa_tagstruct_puts(t, i->sink->name); pa_tagstruct_put_boolean(t, pa_sink_get_state(i->sink) == PA_SINK_SUSPENDED); + + if (s->connection->version >= 13) { + pa_tagstruct_putu32(t, maxlength); + pa_tagstruct_putu32(t, tlength); + pa_tagstruct_putu32(t, prebuf); + pa_tagstruct_putu32(t, minreq); + pa_tagstruct_put_usec(t, s->sink_latency); + } + pa_pstream_send_tagstruct(s->connection->pstream, t); } @@ -1447,11 +1470,19 @@ static void source_output_suspend_cb(pa_source_output *o, pa_bool_t suspend) { static void source_output_moved_cb(pa_source_output *o) { record_stream *s; pa_tagstruct *t; + uint32_t maxlength, fragsize; pa_source_output_assert_ref(o); s = RECORD_STREAM(o->userdata); record_stream_assert_ref(s); + fragsize = (uint32_t) s->fragment_size; + maxlength = (uint32_t) pa_memblockq_get_length(s->memblockq); + + fix_record_buffer_attr_pre(s, TRUE, &maxlength, &fragsize); + pa_memblockq_set_maxlength(s->memblockq, maxlength); + fix_record_buffer_attr_post(s, &maxlength, &fragsize); + if (s->connection->version < 12) return; @@ -1462,6 +1493,13 @@ static void source_output_moved_cb(pa_source_output *o) { pa_tagstruct_putu32(t, o->source->index); pa_tagstruct_puts(t, o->source->name); pa_tagstruct_put_boolean(t, pa_source_get_state(o->source) == PA_SOURCE_SUSPENDED); + + if (s->connection->version >= 13) { + pa_tagstruct_putu32(t, maxlength); + pa_tagstruct_putu32(t, fragsize); + pa_tagstruct_put_usec(t, s->source_latency); + } + pa_pstream_send_tagstruct(s->connection->pstream, t); } @@ -1900,27 +1938,28 @@ static void command_auth(PA_GCC_UNUSED pa_pdispatch *pd, PA_GCC_UNUSED uint32_t pa_proplist_sets(c->client->proplist, "native-protocol.version", tmp); if (!c->authorized) { - int success = 0; + pa_bool_t success = FALSE; #ifdef HAVE_CREDS const pa_creds *creds; if ((creds = pa_pdispatch_creds(pd))) { if (creds->uid == getuid()) - success = 1; + success = TRUE; else if (c->protocol->auth_group) { int r; gid_t gid; if ((gid = pa_get_gid_of_group(c->protocol->auth_group)) == (gid_t) -1) - pa_log_warn("failed to get GID of group '%s'", c->protocol->auth_group); + pa_log_warn("Failed to get GID of group '%s'", c->protocol->auth_group); else if (gid == creds->gid) - success = 1; + success = TRUE; + if (!success) { if ((r = pa_uid_in_group(creds->uid, c->protocol->auth_group)) < 0) - pa_log_warn("failed to check group membership."); + pa_log_warn("Failed to check group membership."); else if (r > 0) - success = 1; + success = TRUE; } } @@ -1941,7 +1980,7 @@ static void command_auth(PA_GCC_UNUSED pa_pdispatch *pd, PA_GCC_UNUSED uint32_t #endif if (!success && memcmp(c->protocol->auth_cookie, cookie, PA_NATIVE_COOKIE_LENGTH) == 0) - success = 1; + success = TRUE; if (!success) { pa_log_warn("Denied access to client with invalid authorization data."); @@ -3037,6 +3076,9 @@ static void command_set_stream_buffer_attr(pa_pdispatch *pd, uint32_t command, u pa_tagstruct_putu32(reply, prebuf); pa_tagstruct_putu32(reply, minreq); + if (c->version >= 13) + pa_tagstruct_put_usec(reply, s->sink_latency); + } else { record_stream *s; pa_bool_t adjust_latency = FALSE; @@ -3063,6 +3105,9 @@ static void command_set_stream_buffer_attr(pa_pdispatch *pd, uint32_t command, u reply = reply_new(tag); pa_tagstruct_putu32(reply, maxlength); pa_tagstruct_putu32(reply, fragsize); + + if (c->version >= 13) + pa_tagstruct_put_usec(reply, s->source_latency); } pa_pstream_send_tagstruct(c->pstream, reply); @@ -3600,7 +3645,7 @@ static void command_move_stream(pa_pdispatch *pd, uint32_t command, uint32_t tag CHECK_VALIDITY(c->pstream, si && sink, tag, PA_ERR_NOENTITY); - if (pa_sink_input_move_to(si, sink, 0) < 0) { + if (pa_sink_input_move_to(si, sink) < 0) { pa_pstream_send_error(c->pstream, tag, PA_ERR_INVALID); return; } diff --git a/src/pulsecore/sink-input.c b/src/pulsecore/sink-input.c index 0be1cc97..d51ff810 100644 --- a/src/pulsecore/sink-input.c +++ b/src/pulsecore/sink-input.c @@ -243,7 +243,6 @@ pa_sink_input* pa_sink_input_new( i->thread_info.state = i->state; i->thread_info.attached = FALSE; pa_atomic_store(&i->thread_info.drained, 1); - pa_atomic_store(&i->thread_info.render_memblockq_is_empty, 0); i->thread_info.sample_spec = i->sample_spec; i->thread_info.resampler = resampler; i->thread_info.volume = i->volume; @@ -410,10 +409,6 @@ void pa_sink_input_put(pa_sink_input *i) { pa_subscription_post(i->sink->core, PA_SUBSCRIPTION_EVENT_SINK_INPUT|PA_SUBSCRIPTION_EVENT_NEW, i->index); pa_hook_fire(&i->sink->core->hooks[PA_CORE_HOOK_SINK_INPUT_PUT], i); - - /* Please note that if you change something here, you have to - change something in pa_sink_input_move() with the ghost stream - registration too. */ } void pa_sink_input_kill(pa_sink_input*i) { @@ -511,10 +506,10 @@ int pa_sink_input_peek(pa_sink_input *i, size_t slength /* in sink frames */, pa * data, so let's just hand out silence */ pa_atomic_store(&i->thread_info.drained, 1); - pa_memblockq_seek(i->thread_info.render_memblockq, slength, PA_SEEK_RELATIVE_ON_READ); + pa_memblockq_seek(i->thread_info.render_memblockq, slength, PA_SEEK_RELATIVE); i->thread_info.playing_for = 0; if (i->thread_info.underrun_for != (uint64_t) -1) - i->thread_info.underrun_for += slength; + i->thread_info.underrun_for += ilength; break; } @@ -551,6 +546,8 @@ int pa_sink_input_peek(pa_sink_input *i, size_t slength /* in sink frames */, pa pa_memchunk rchunk; pa_resampler_run(i->thread_info.resampler, &wchunk, &rchunk); +/* pa_log_debug("pushing %lu", (unsigned long) rchunk.length); */ + if (rchunk.memblock) { pa_memblockq_push_align(i->thread_info.render_memblockq, &rchunk); pa_memblock_unref(rchunk.memblock); @@ -571,6 +568,8 @@ int pa_sink_input_peek(pa_sink_input *i, size_t slength /* in sink frames */, pa pa_assert(chunk->length > 0); pa_assert(chunk->memblock); +/* pa_log_debug("peeking %lu", (unsigned long) chunk->length); */ + if (chunk->length > block_size_max_sink) chunk->length = block_size_max_sink; @@ -586,8 +585,6 @@ int pa_sink_input_peek(pa_sink_input *i, size_t slength /* in sink frames */, pa else *volume = i->thread_info.volume; - pa_atomic_store(&i->thread_info.render_memblockq_is_empty, pa_memblockq_is_empty(i->thread_info.render_memblockq)); - return 0; } @@ -599,6 +596,8 @@ void pa_sink_input_drop(pa_sink_input *i, size_t nbytes /* in sink sample spec * pa_assert(pa_frame_aligned(nbytes, &i->sink->sample_spec)); pa_assert(nbytes > 0); +/* pa_log_debug("dropping %lu", (unsigned long) nbytes); */ + /* If there's still some rewrite request the handle, but the sink didn't do this for us, we do it here. However, since the sink apparently doesn't support rewinding, we pass 0 here. This still @@ -606,12 +605,11 @@ void pa_sink_input_drop(pa_sink_input *i, size_t nbytes /* in sink sample spec * pa_sink_input_process_rewind(i, 0); pa_memblockq_drop(i->thread_info.render_memblockq, nbytes); - - pa_atomic_store(&i->thread_info.render_memblockq_is_empty, pa_memblockq_is_empty(i->thread_info.render_memblockq)); } /* Called from thread context */ void pa_sink_input_process_rewind(pa_sink_input *i, size_t nbytes /* in sink sample spec */) { + size_t lbq; pa_sink_input_assert_ref(i); pa_assert(PA_SINK_INPUT_IS_LINKED(i->thread_info.state)); @@ -619,6 +617,8 @@ void pa_sink_input_process_rewind(pa_sink_input *i, size_t nbytes /* in sink sam /* pa_log_debug("rewind(%lu, %lu)", (unsigned long) nbytes, (unsigned long) i->thread_info.rewrite_nbytes); */ + lbq = pa_memblockq_get_length(i->thread_info.render_memblockq); + if (nbytes > 0) { pa_log_debug("Have to rewind %lu bytes on render memblockq.", (unsigned long) nbytes); pa_memblockq_rewind(i->thread_info.render_memblockq, nbytes); @@ -635,7 +635,7 @@ void pa_sink_input_process_rewind(pa_sink_input *i, size_t nbytes /* in sink sam size_t max_rewrite, amount; /* Calculate how much make sense to rewrite at most */ - max_rewrite = nbytes + pa_memblockq_get_length(i->thread_info.render_memblockq); + max_rewrite = nbytes + lbq; /* Transform into local domain */ if (i->thread_info.resampler) @@ -651,18 +651,16 @@ void pa_sink_input_process_rewind(pa_sink_input *i, size_t nbytes /* in sink sam if (i->process_rewind) i->process_rewind(i, amount); - if (i->thread_info.rewrite_flush) - pa_memblockq_silence(i->thread_info.render_memblockq); - else { + /* Convert back to to sink domain */ + if (i->thread_info.resampler) + amount = pa_resampler_result(i->thread_info.resampler, amount); - /* Convert back to to sink domain */ - if (i->thread_info.resampler) - amount = pa_resampler_result(i->thread_info.resampler, amount); + if (amount > 0) + /* Ok, now update the write pointer */ + pa_memblockq_seek(i->thread_info.render_memblockq, - ((int64_t) amount), PA_SEEK_RELATIVE); - if (amount > 0) - /* Ok, now update the write pointer */ - pa_memblockq_seek(i->thread_info.render_memblockq, - ((int64_t) amount), PA_SEEK_RELATIVE); - } + if (i->thread_info.rewrite_flush) + pa_memblockq_silence(i->thread_info.render_memblockq); /* And reset the resampler */ if (i->thread_info.resampler) @@ -702,6 +700,7 @@ static pa_usec_t fixup_latency(pa_sink *s, pa_usec_t usec) { } pa_usec_t pa_sink_input_set_requested_latency_within_thread(pa_sink_input *i, pa_usec_t usec) { + pa_sink_input_assert_ref(i); usec = fixup_latency(i->sink, usec); @@ -728,6 +727,21 @@ pa_usec_t pa_sink_input_set_requested_latency(pa_sink_input *i, pa_usec_t usec) return usec; } +pa_usec_t pa_sink_input_get_requested_latency(pa_sink_input *i) { + pa_usec_t usec = 0; + + pa_sink_input_assert_ref(i); + + if (PA_SINK_INPUT_IS_LINKED(i->state)) + pa_asyncmsgq_send(i->sink->asyncmsgq, PA_MSGOBJECT(i), PA_SINK_INPUT_MESSAGE_GET_REQUESTED_LATENCY, &usec, 0, NULL); + else + /* If this sink input is not realized yet, we have to touch + * the thread info data directly */ + usec = i->thread_info.requested_sink_latency; + + return usec; +} + void pa_sink_input_set_volume(pa_sink_input *i, const pa_cvolume *volume) { pa_sink_input_assert_ref(i); pa_assert(PA_SINK_INPUT_IS_LINKED(i->state)); @@ -821,11 +835,9 @@ pa_resample_method_t pa_sink_input_get_resample_method(pa_sink_input *i) { return i->resample_method; } -int pa_sink_input_move_to(pa_sink_input *i, pa_sink *dest, pa_bool_t immediately) { +int pa_sink_input_move_to(pa_sink_input *i, pa_sink *dest) { pa_resampler *new_resampler; pa_sink *origin; - pa_usec_t silence_usec = 0; - pa_sink_input_move_info info; pa_sink_input_move_hook_data hook_data; pa_sink_input_assert_ref(i); @@ -881,80 +893,7 @@ int pa_sink_input_move_to(pa_sink_input *i, pa_sink *dest, pa_bool_t immediately hook_data.destination = dest; pa_hook_fire(&i->sink->core->hooks[PA_CORE_HOOK_SINK_INPUT_MOVE], &hook_data); - memset(&info, 0, sizeof(info)); - info.sink_input = i; - - if (!immediately) { - pa_usec_t old_latency, new_latency; - - /* Let's do a little bit of Voodoo for compensating latency - * differences. We assume that the accuracy for our - * estimations is still good enough, even though we do these - * operations non-atomic. */ - - old_latency = pa_sink_get_latency(origin); - new_latency = pa_sink_get_latency(dest); - - /* The already resampled data should go to the old sink */ - - if (old_latency >= new_latency) { - - /* The latency of the old sink is larger than the latency - * of the new sink. Therefore to compensate for the - * difference we to play silence on the new one for a - * while */ - - silence_usec = old_latency - new_latency; - - } else { - - /* The latency of new sink is larger than the latency of - * the old sink. Therefore we have to precompute a little - * and make sure that this is still played on the old - * sink, until we can play the first sample on the new - * sink.*/ - - info.buffer_bytes = pa_usec_to_bytes(new_latency - old_latency, &origin->sample_spec); - } - - /* Okey, let's move it */ - - if (info.buffer_bytes > 0) { - pa_proplist *p; - - p = pa_proplist_new(); - pa_proplist_sets(p, PA_PROP_MEDIA_NAME, "Ghost For Moved Stream"); - pa_proplist_sets(p, PA_PROP_MEDIA_ROLE, "routing"); - - info.ghost_sink_input = pa_memblockq_sink_input_new( - origin, - &origin->sample_spec, - &origin->channel_map, - NULL, - NULL, - p); - - pa_proplist_free(p); - - if (info.ghost_sink_input) { - info.ghost_sink_input->thread_info.state = info.ghost_sink_input->state = PA_SINK_INPUT_RUNNING; - info.ghost_sink_input->thread_info.volume = info.ghost_sink_input->volume; - info.ghost_sink_input->thread_info.muted = info.ghost_sink_input->muted; - - info.buffer = pa_memblockq_new(0, MOVE_BUFFER_LENGTH, 0, pa_frame_size(&origin->sample_spec), 0, 0, 0, NULL); - } - } - } - - pa_asyncmsgq_send(i->sink->asyncmsgq, PA_MSGOBJECT(i->sink), PA_SINK_MESSAGE_REMOVE_INPUT_AND_BUFFER, &info, 0, NULL); - - if (info.ghost_sink_input) { - /* Basically, do what pa_sink_input_put() does ...*/ - - pa_subscription_post(i->sink->core, PA_SUBSCRIPTION_EVENT_SINK_INPUT|PA_SUBSCRIPTION_EVENT_NEW, info.ghost_sink_input->index); - pa_hook_fire(&i->sink->core->hooks[PA_CORE_HOOK_SINK_INPUT_PUT], info.ghost_sink_input); - pa_sink_input_unref(info.ghost_sink_input); - } + pa_asyncmsgq_send(i->sink->asyncmsgq, PA_MSGOBJECT(i->sink), PA_SINK_MESSAGE_START_MOVE, i, 0, NULL); pa_idxset_remove_by_data(origin->inputs, i, NULL); pa_idxset_put(dest->inputs, i, NULL); @@ -965,40 +904,31 @@ int pa_sink_input_move_to(pa_sink_input *i, pa_sink *dest, pa_bool_t immediately dest->n_corked++; } - /* Replace resampler */ + /* Replace resampler and render queue */ if (new_resampler != i->thread_info.resampler) { - pa_memchunk silence; if (i->thread_info.resampler) pa_resampler_free(i->thread_info.resampler); i->thread_info.resampler = new_resampler; - /* if the resampler changed, the silence memblock is - * probably invalid now, too */ - - pa_silence_memchunk_get( - &i->sink->core->silence_cache, - i->sink->core->mempool, - &silence, - &dest->sample_spec, - 0); - - pa_memblockq_set_silence(i->thread_info.render_memblockq, &silence); - pa_memblock_unref(silence.memblock); + pa_memblockq_free(i->thread_info.render_memblockq); + i->thread_info.render_memblockq = pa_memblockq_new( + 0, + MEMBLOCKQ_MAXLENGTH, + 0, + pa_frame_size(&i->sink->sample_spec), + 0, + 1, + 0, + &i->sink->silence); } - pa_memblockq_flush(i->thread_info.render_memblockq); - - /* Calculate the new sleeping time */ - if (!immediately) - pa_memblockq_seek(i->thread_info.render_memblockq, pa_usec_to_bytes(silence_usec, &dest->sample_spec), PA_SEEK_RELATIVE); - - pa_asyncmsgq_send(i->sink->asyncmsgq, PA_MSGOBJECT(i->sink), PA_SINK_MESSAGE_ADD_INPUT, i, 0, NULL); - pa_sink_update_status(origin); pa_sink_update_status(dest); + pa_asyncmsgq_send(i->sink->asyncmsgq, PA_MSGOBJECT(i->sink), PA_SINK_MESSAGE_FINISH_MOVE, i, 0, NULL); + if (i->moved) i->moved(i); @@ -1015,6 +945,9 @@ int pa_sink_input_move_to(pa_sink_input *i, pa_sink *dest, pa_bool_t immediately void pa_sink_input_set_state_within_thread(pa_sink_input *i, pa_sink_input_state_t state) { pa_sink_input_assert_ref(i); + if (state == i->thread_info.state) + return; + if ((state == PA_SINK_INPUT_DRAINED || state == PA_SINK_INPUT_RUNNING) && !(i->thread_info.state == PA_SINK_INPUT_DRAINED || i->thread_info.state != PA_SINK_INPUT_RUNNING)) pa_atomic_store(&i->thread_info.drained, 1); @@ -1060,7 +993,6 @@ int pa_sink_input_process_msg(pa_msgobject *o, int code, void *userdata, int64_t pa_usec_t *r = userdata; *r += pa_bytes_to_usec(pa_memblockq_get_length(i->thread_info.render_memblockq), &i->sink->sample_spec); - return 0; } @@ -1089,6 +1021,13 @@ int pa_sink_input_process_msg(pa_msgobject *o, int code, void *userdata, int64_t pa_sink_input_set_requested_latency_within_thread(i, (pa_usec_t) offset); return 0; + + case PA_SINK_INPUT_MESSAGE_GET_REQUESTED_LATENCY: { + pa_usec_t *r = userdata; + + *r = i->thread_info.requested_sink_latency; + return 0; + } } return -1; @@ -1103,11 +1042,12 @@ pa_sink_input_state_t pa_sink_input_get_state(pa_sink_input *i) { return i->state; } +/* Called from IO context */ pa_bool_t pa_sink_input_safe_to_remove(pa_sink_input *i) { pa_sink_input_assert_ref(i); - if (i->state == PA_SINK_INPUT_RUNNING || i->state == PA_SINK_INPUT_DRAINED || i->state == PA_SINK_INPUT_CORKED) - return pa_atomic_load(&i->thread_info.render_memblockq_is_empty); + if (PA_SINK_INPUT_IS_LINKED(i->thread_info.state)) + return pa_memblockq_is_empty(i->thread_info.render_memblockq); return TRUE; } @@ -1129,7 +1069,7 @@ void pa_sink_input_request_rewind(pa_sink_input *i, size_t nbytes /* in our sam pa_assert(i->thread_info.rewrite_nbytes == 0); /* We don't take rewind requests while we are corked */ - if (i->state == PA_SINK_INPUT_CORKED) + if (i->thread_info.state == PA_SINK_INPUT_CORKED) return; pa_assert(rewrite || flush); diff --git a/src/pulsecore/sink-input.h b/src/pulsecore/sink-input.h index 8edd7ecb..5f146122 100644 --- a/src/pulsecore/sink-input.h +++ b/src/pulsecore/sink-input.h @@ -144,7 +144,7 @@ struct pa_sink_input { struct { pa_sink_input_state_t state; - pa_atomic_t drained, render_memblockq_is_empty; + pa_atomic_t drained; pa_bool_t attached; /* True only between ->attach() and ->detach() calls */ @@ -181,6 +181,7 @@ enum { PA_SINK_INPUT_MESSAGE_SET_RATE, PA_SINK_INPUT_MESSAGE_SET_STATE, PA_SINK_INPUT_MESSAGE_SET_REQUESTED_LATENCY, + PA_SINK_INPUT_MESSAGE_GET_REQUESTED_LATENCY, PA_SINK_INPUT_MESSAGE_MAX }; @@ -243,6 +244,10 @@ could be rewound in the HW device. This functionality is required for implementing the "zero latency" write-through functionality. */ void pa_sink_input_request_rewind(pa_sink_input *i, size_t nbytes, pa_bool_t rewrite, pa_bool_t flush); +void pa_sink_input_cork(pa_sink_input *i, pa_bool_t b); + +int pa_sink_input_set_rate(pa_sink_input *i, uint32_t rate); + /* Callable by everyone from main thread*/ /* External code may request disconnection with this function */ @@ -255,17 +260,14 @@ const pa_cvolume *pa_sink_input_get_volume(pa_sink_input *i); void pa_sink_input_set_mute(pa_sink_input *i, pa_bool_t mute); int pa_sink_input_get_mute(pa_sink_input *i); -void pa_sink_input_cork(pa_sink_input *i, pa_bool_t b); - -int pa_sink_input_set_rate(pa_sink_input *i, uint32_t rate); - pa_resample_method_t pa_sink_input_get_resample_method(pa_sink_input *i); -int pa_sink_input_move_to(pa_sink_input *i, pa_sink *dest, pa_bool_t immediately); +int pa_sink_input_move_to(pa_sink_input *i, pa_sink *dest); pa_sink_input_state_t pa_sink_input_get_state(pa_sink_input *i); -pa_bool_t pa_sink_input_safe_to_remove(pa_sink_input *i); +pa_usec_t pa_sink_input_get_requested_latency(pa_sink_input *i); + /* To be used exclusively by the sink driver IO thread */ int pa_sink_input_peek(pa_sink_input *i, size_t length, pa_memchunk *chunk, pa_cvolume *volume); @@ -279,12 +281,7 @@ int pa_sink_input_process_msg(pa_msgobject *o, int code, void *userdata, int64_t pa_usec_t pa_sink_input_set_requested_latency_within_thread(pa_sink_input *i, pa_usec_t usec); -typedef struct pa_sink_input_move_info { - pa_sink_input *sink_input; - pa_sink_input *ghost_sink_input; - pa_memblockq *buffer; - size_t buffer_bytes; -} pa_sink_input_move_info; +pa_bool_t pa_sink_input_safe_to_remove(pa_sink_input *i); pa_memchunk* pa_sink_input_get_silence(pa_sink_input *i, pa_memchunk *ret); diff --git a/src/pulsecore/sink.c b/src/pulsecore/sink.c index d3bacbfd..31c3cfc8 100644 --- a/src/pulsecore/sink.c +++ b/src/pulsecore/sink.c @@ -113,7 +113,6 @@ static void reset_callbacks(pa_sink *s) { s->set_volume = NULL; s->get_mute = NULL; s->set_mute = NULL; - s->get_latency = NULL; s->request_rewind = NULL; s->update_requested_latency = NULL; } @@ -769,9 +768,6 @@ pa_usec_t pa_sink_get_latency(pa_sink *s) { if (!PA_SINK_IS_OPENED(s->state)) return 0; - if (s->get_latency) - return s->get_latency(s); - if (pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SINK_MESSAGE_GET_LATENCY, &usec, 0, NULL) < 0) return 0; @@ -930,6 +926,10 @@ int pa_sink_process_msg(pa_msgobject *o, int code, void *userdata, int64_t offse case PA_SINK_MESSAGE_ADD_INPUT: { pa_sink_input *i = PA_SINK_INPUT(userdata); + /* If you change anything here, make sure to change the + * sink input handling a few lines down at + * PA_SINK_MESSAGE_FINISH_MOVE, too. */ + pa_hashmap_put(s->thread_info.inputs, PA_UINT32_TO_PTR(i->index), pa_sink_input_ref(i)); /* Since the caller sleeps in pa_sink_input_put(), we can @@ -965,10 +965,6 @@ int pa_sink_process_msg(pa_msgobject *o, int code, void *userdata, int64_t offse * slow start, i.e. need some time to buffer client * samples before beginning streaming. */ - /* If you change anything here, make sure to change the - * ghost sink input handling a few lines down at - * PA_SINK_MESSAGE_REMOVE_INPUT_AND_BUFFER, too. */ - return 0; } @@ -977,7 +973,7 @@ int pa_sink_process_msg(pa_msgobject *o, int code, void *userdata, int64_t offse /* If you change anything here, make sure to change the * sink input handling a few lines down at - * PA_SINK_MESSAGE_REMOVE_INPUT_AND_BUFFER, too. */ + * PA_SINK_MESSAGE_PREPAPRE_MOVE, too. */ pa_sink_input_set_state_within_thread(i, i->state); @@ -1013,85 +1009,88 @@ int pa_sink_process_msg(pa_msgobject *o, int code, void *userdata, int64_t offse return 0; } - case PA_SINK_MESSAGE_REMOVE_INPUT_AND_BUFFER: { - pa_sink_input_move_info *info = userdata; - int volume_is_norm; + case PA_SINK_MESSAGE_START_MOVE: { + pa_sink_input *i = PA_SINK_INPUT(userdata); /* We don't support moving synchronized streams. */ - pa_assert(!info->sink_input->sync_prev); - pa_assert(!info->sink_input->sync_next); - pa_assert(!info->sink_input->thread_info.sync_next); - pa_assert(!info->sink_input->thread_info.sync_prev); + pa_assert(!i->sync_prev); + pa_assert(!i->sync_next); + pa_assert(!i->thread_info.sync_next); + pa_assert(!i->thread_info.sync_prev); - if (info->sink_input->detach) - info->sink_input->detach(info->sink_input); + if (i->thread_info.state != PA_SINK_INPUT_CORKED) { + pa_usec_t usec = 0; + size_t sink_nbytes, total_nbytes; - pa_assert(info->sink_input->thread_info.attached); - info->sink_input->thread_info.attached = FALSE; - pa_sink_invalidate_requested_latency(info->sink_input->sink); + /* Get the latency of the sink */ + if (PA_MSGOBJECT(s)->process_msg(PA_MSGOBJECT(s), PA_SINK_MESSAGE_GET_LATENCY, &usec, 0, NULL) < 0) + usec = 0; - if (info->ghost_sink_input) { - pa_assert(info->buffer_bytes > 0); - pa_assert(info->buffer); + sink_nbytes = pa_usec_to_bytes(usec, &s->sample_spec); + total_nbytes = sink_nbytes + pa_memblockq_get_length(i->thread_info.render_memblockq); - volume_is_norm = pa_cvolume_is_norm(&info->sink_input->thread_info.volume); + if (total_nbytes > 0) { + i->thread_info.rewrite_nbytes = i->thread_info.resampler ? pa_resampler_request(i->thread_info.resampler, total_nbytes) : total_nbytes; + i->thread_info.rewrite_flush = TRUE; + pa_sink_input_process_rewind(i, sink_nbytes); + } + } - pa_log_debug("Buffering %lu bytes ...", (unsigned long) info->buffer_bytes); + if (i->detach) + i->detach(i); - while (info->buffer_bytes > 0) { - pa_memchunk memchunk; - pa_cvolume volume; - size_t n; + pa_assert(i->thread_info.attached); + i->thread_info.attached = FALSE; - if (pa_sink_input_peek(info->sink_input, info->buffer_bytes, &memchunk, &volume) < 0) - break; + /* Let's remove the sink input ...*/ + if (pa_hashmap_remove(s->thread_info.inputs, PA_UINT32_TO_PTR(i->index))) + pa_sink_input_unref(i); - n = memchunk.length > info->buffer_bytes ? info->buffer_bytes : memchunk.length; - pa_sink_input_drop(info->sink_input, n); - memchunk.length = n; + pa_sink_invalidate_requested_latency(s); - if (!volume_is_norm) { - pa_memchunk_make_writable(&memchunk, 0); - pa_volume_memchunk(&memchunk, &s->sample_spec, &volume); - } + pa_log_debug("Requesting rewind due to started move"); + pa_sink_request_rewind(s, 0); - if (pa_memblockq_push(info->buffer, &memchunk) < 0) { - pa_memblock_unref(memchunk.memblock); - break; - } + return 0; + } - pa_memblock_unref(memchunk.memblock); - info->buffer_bytes -= n; - } + case PA_SINK_MESSAGE_FINISH_MOVE: { + pa_sink_input *i = PA_SINK_INPUT(userdata); - /* Add the remaining already resampled chunks to the buffer */ - pa_memblockq_splice(info->buffer, info->sink_input->thread_info.render_memblockq); + /* We don't support moving synchronized streams. */ + pa_assert(!i->sync_prev); + pa_assert(!i->sync_next); + pa_assert(!i->thread_info.sync_next); + pa_assert(!i->thread_info.sync_prev); - pa_memblockq_sink_input_set_queue(info->ghost_sink_input, info->buffer); + pa_hashmap_put(s->thread_info.inputs, PA_UINT32_TO_PTR(i->index), pa_sink_input_ref(i)); - pa_log_debug("Buffered %lu bytes ...", (unsigned long) pa_memblockq_get_length(info->buffer)); - } + pa_assert(!i->thread_info.attached); + i->thread_info.attached = TRUE; - /* Let's remove the sink input ...*/ - if (pa_hashmap_remove(s->thread_info.inputs, PA_UINT32_TO_PTR(info->sink_input->index))) - pa_sink_input_unref(info->sink_input); + if (i->attach) + i->attach(i); - /* .. and add the ghost sink input instead */ - if (info->ghost_sink_input) { - pa_hashmap_put(s->thread_info.inputs, PA_UINT32_TO_PTR(info->ghost_sink_input->index), pa_sink_input_ref(info->ghost_sink_input)); - info->ghost_sink_input->thread_info.sync_prev = info->ghost_sink_input->thread_info.sync_next = NULL; + pa_sink_input_update_max_rewind(i, s->thread_info.max_rewind); - pa_sink_input_update_max_rewind(info->ghost_sink_input, s->thread_info.max_rewind); + pa_sink_input_set_requested_latency_within_thread(i, i->thread_info.requested_sink_latency); - pa_assert(!info->ghost_sink_input->thread_info.attached); - info->ghost_sink_input->thread_info.attached = TRUE; + if (i->thread_info.state != PA_SINK_INPUT_CORKED) { + pa_usec_t usec = 0; + size_t nbytes; - if (info->ghost_sink_input->attach) - info->ghost_sink_input->attach(info->ghost_sink_input); - } + /* Get the latency of the sink */ + if (PA_MSGOBJECT(s)->process_msg(PA_MSGOBJECT(s), PA_SINK_MESSAGE_GET_LATENCY, &usec, 0, NULL) < 0) + usec = 0; - pa_sink_invalidate_requested_latency(s); - pa_sink_request_rewind(s, 0); + nbytes = pa_usec_to_bytes(usec, &s->sample_spec); + + if (nbytes > 0) + pa_sink_input_drop(i, nbytes); + + pa_log_debug("Requesting rewind due to finished move"); + pa_sink_request_rewind(s, nbytes); + } return 0; } diff --git a/src/pulsecore/sink.h b/src/pulsecore/sink.h index f25f48cf..f297c8f1 100644 --- a/src/pulsecore/sink.h +++ b/src/pulsecore/sink.h @@ -118,11 +118,6 @@ struct pa_sink { * message will be sent to the IO thread instead. */ int (*set_mute)(pa_sink *s); /* dito */ - /* Called when the latency is queried. Called from main loop - context. If this is NULL a PA_SINK_MESSAGE_GET_LATENCY message - will be sent to the IO thread instead. */ - pa_usec_t (*get_latency)(pa_sink *s); /* dito */ - /* Called when a rewind request is issued. Called from IO thread * context. */ void (*request_rewind)(pa_sink *s); /* dito */ @@ -166,7 +161,8 @@ typedef enum pa_sink_message { PA_SINK_MESSAGE_GET_LATENCY, PA_SINK_MESSAGE_GET_REQUESTED_LATENCY, PA_SINK_MESSAGE_SET_STATE, - PA_SINK_MESSAGE_REMOVE_INPUT_AND_BUFFER, + PA_SINK_MESSAGE_START_MOVE, + PA_SINK_MESSAGE_FINISH_MOVE, PA_SINK_MESSAGE_ATTACH, PA_SINK_MESSAGE_DETACH, PA_SINK_MESSAGE_MAX diff --git a/src/pulsecore/source-output.c b/src/pulsecore/source-output.c index 836e30ed..3940d768 100644 --- a/src/pulsecore/source-output.c +++ b/src/pulsecore/source-output.c @@ -294,6 +294,7 @@ void pa_source_output_unlink(pa_source_output*o) { static void source_output_free(pa_object* mo) { pa_source_output *o = PA_SOURCE_OUTPUT(mo); + pa_assert(o); pa_assert(pa_source_output_refcnt(o) == 0); if (PA_SOURCE_OUTPUT_IS_LINKED(o->state)) @@ -326,7 +327,7 @@ void pa_source_output_put(pa_source_output *o) { state = o->flags & PA_SOURCE_OUTPUT_START_CORKED ? PA_SOURCE_OUTPUT_CORKED : PA_SOURCE_OUTPUT_RUNNING; update_n_corked(o, state); - o->thread_info.state = o->state = state; + o->state = state; pa_asyncmsgq_send(o->source->asyncmsgq, PA_MSGOBJECT(o->source), PA_SOURCE_MESSAGE_ADD_OUTPUT, o, 0, NULL); @@ -470,6 +471,7 @@ static pa_usec_t fixup_latency(pa_source *s, pa_usec_t usec) { } pa_usec_t pa_source_output_set_requested_latency_within_thread(pa_source_output *o, pa_usec_t usec) { + pa_source_output_assert_ref(o); usec = fixup_latency(o->source, usec); @@ -496,6 +498,21 @@ pa_usec_t pa_source_output_set_requested_latency(pa_source_output *o, pa_usec_t return usec; } +pa_usec_t pa_source_output_get_requested_latency(pa_source_output *o) { + pa_usec_t usec = 0; + + pa_source_output_assert_ref(o); + + if (PA_SOURCE_OUTPUT_IS_LINKED(o->state)) + pa_asyncmsgq_send(o->source->asyncmsgq, PA_MSGOBJECT(o), PA_SOURCE_OUTPUT_MESSAGE_GET_REQUESTED_LATENCY, &usec, 0, NULL); + else + /* If this sink input is not realized yet, we have to touch + * the thread info data directly */ + usec = o->thread_info.requested_source_latency; + + return usec; +} + void pa_source_output_cork(pa_source_output *o, pa_bool_t b) { pa_source_output_assert_ref(o); pa_assert(PA_SOURCE_OUTPUT_IS_LINKED(o->state)); @@ -523,11 +540,11 @@ void pa_source_output_set_name(pa_source_output *o, const char *name) { const char *old; pa_source_output_assert_ref(o); - old = pa_proplist_gets(o->proplist, PA_PROP_MEDIA_NAME); - - if (!old && !name) + if (!name && !pa_proplist_contains(o->proplist, PA_PROP_MEDIA_NAME)) return; + old = pa_proplist_gets(o->proplist, PA_PROP_MEDIA_NAME); + if (old && name && !strcmp(old, name)) return; @@ -550,7 +567,7 @@ pa_resample_method_t pa_source_output_get_resample_method(pa_source_output *o) { int pa_source_output_move_to(pa_source_output *o, pa_source *dest) { pa_source *origin; - pa_resampler *new_resampler = NULL; + pa_resampler *new_resampler; pa_source_output_move_hook_data hook_data; pa_source_output_assert_ref(o); @@ -594,7 +611,8 @@ int pa_source_output_move_to(pa_source_output *o, pa_source *dest) { pa_log_warn("Unsupported resampling operation."); return -1; } - } + } else + new_resampler = NULL; hook_data.source_output = o; hook_data.destination = dest; @@ -640,6 +658,9 @@ int pa_source_output_move_to(pa_source_output *o, pa_source *dest) { void pa_source_output_set_state_within_thread(pa_source_output *o, pa_source_output_state_t state) { pa_source_output_assert_ref(o); + if (state == o->thread_info.state) + return; + if (o->state_change) o->state_change(o, state); @@ -659,7 +680,6 @@ int pa_source_output_process_msg(pa_msgobject *mo, int code, void *userdata, int pa_usec_t *r = userdata; *r += pa_bytes_to_usec(pa_memblockq_get_length(o->thread_info.delay_memblockq), &o->source->sample_spec); - return 0; } @@ -678,6 +698,13 @@ int pa_source_output_process_msg(pa_msgobject *mo, int code, void *userdata, int pa_source_output_set_requested_latency_within_thread(o, (pa_usec_t) offset); return 0; + + case PA_SINK_INPUT_MESSAGE_GET_REQUESTED_LATENCY: { + pa_usec_t *r = userdata; + + *r = o->thread_info.requested_source_latency; + return 0; + } } return -1; diff --git a/src/pulsecore/source-output.h b/src/pulsecore/source-output.h index 67cb3761..2dadb5c4 100644 --- a/src/pulsecore/source-output.h +++ b/src/pulsecore/source-output.h @@ -148,6 +148,7 @@ enum { PA_SOURCE_OUTPUT_MESSAGE_SET_RATE, PA_SOURCE_OUTPUT_MESSAGE_SET_STATE, PA_SOURCE_OUTPUT_MESSAGE_SET_REQUESTED_LATENCY, + PA_SOURCE_OUTPUT_MESSAGE_GET_REQUESTED_LATENCY, PA_SOURCE_OUTPUT_MESSAGE_MAX }; @@ -168,16 +169,16 @@ typedef struct pa_source_output_new_data { pa_resample_method_t resample_method; } pa_source_output_new_data; -typedef struct pa_source_output_move_hook_data { - pa_source_output *source_output; - pa_source *destination; -} pa_source_output_move_hook_data; - pa_source_output_new_data* pa_source_output_new_data_init(pa_source_output_new_data *data); void pa_source_output_new_data_set_sample_spec(pa_source_output_new_data *data, const pa_sample_spec *spec); void pa_source_output_new_data_set_channel_map(pa_source_output_new_data *data, const pa_channel_map *map); void pa_source_output_new_data_done(pa_source_output_new_data *data); +typedef struct pa_source_output_move_hook_data { + pa_source_output *source_output; + pa_source *destination; +} pa_source_output_move_hook_data; + /* To be called by the implementing module only */ pa_source_output* pa_source_output_new( @@ -192,6 +193,10 @@ void pa_source_output_set_name(pa_source_output *i, const char *name); pa_usec_t pa_source_output_set_requested_latency(pa_source_output *i, pa_usec_t usec); +void pa_source_output_cork(pa_source_output *i, pa_bool_t b); + +int pa_source_output_set_rate(pa_source_output *o, uint32_t rate); + /* Callable by everyone */ /* External code may request disconnection with this funcion */ @@ -199,26 +204,24 @@ void pa_source_output_kill(pa_source_output*o); pa_usec_t pa_source_output_get_latency(pa_source_output *i); -void pa_source_output_cork(pa_source_output *i, pa_bool_t b); - -int pa_source_output_set_rate(pa_source_output *o, uint32_t rate); - pa_resample_method_t pa_source_output_get_resample_method(pa_source_output *o); int pa_source_output_move_to(pa_source_output *o, pa_source *dest); #define pa_source_output_get_state(o) ((o)->state) +pa_usec_t pa_source_output_get_requested_latency(pa_source_output *o); + /* To be used exclusively by the source driver thread */ void pa_source_output_push(pa_source_output *o, const pa_memchunk *chunk); void pa_source_output_process_rewind(pa_source_output *o, size_t nbytes); void pa_source_output_update_max_rewind(pa_source_output *o, size_t nbytes); -int pa_source_output_process_msg(pa_msgobject *mo, int code, void *userdata, int64_t offset, pa_memchunk *chunk); - void pa_source_output_set_state_within_thread(pa_source_output *o, pa_source_output_state_t state); +int pa_source_output_process_msg(pa_msgobject *mo, int code, void *userdata, int64_t offset, pa_memchunk *chunk); + pa_usec_t pa_source_output_set_requested_latency_within_thread(pa_source_output *o, pa_usec_t usec); #endif diff --git a/src/pulsecore/source.c b/src/pulsecore/source.c index fc4734fd..426906eb 100644 --- a/src/pulsecore/source.c +++ b/src/pulsecore/source.c @@ -107,7 +107,6 @@ static void reset_callbacks(pa_source *s) { s->set_volume = NULL; s->get_mute = NULL; s->set_mute = NULL; - s->get_latency = NULL; s->update_requested_latency = NULL; } @@ -440,9 +439,6 @@ pa_usec_t pa_source_get_latency(pa_source *s) { if (!PA_SOURCE_IS_OPENED(s->state)) return 0; - if (s->get_latency) - return s->get_latency(s); - if (pa_asyncmsgq_send(s->asyncmsgq, PA_MSGOBJECT(s), PA_SOURCE_MESSAGE_GET_LATENCY, &usec, 0, NULL) < 0) return 0; diff --git a/src/pulsecore/source.h b/src/pulsecore/source.h index cce54620..f9c9cbf9 100644 --- a/src/pulsecore/source.h +++ b/src/pulsecore/source.h @@ -100,7 +100,6 @@ struct pa_source { int (*get_volume)(pa_source *s); /* dito */ int (*set_mute)(pa_source *s); /* dito */ int (*get_mute)(pa_source *s); /* dito */ - pa_usec_t (*get_latency)(pa_source *s); /* dito */ void (*update_requested_latency)(pa_source *s); /* dito */ /* Contains copies of the above data so that the real-time worker |