summaryrefslogtreecommitdiffstats
path: root/src/modules
diff options
context:
space:
mode:
Diffstat (limited to 'src/modules')
-rw-r--r--src/modules/alsa/alsa-sink.c2
-rw-r--r--src/modules/alsa/alsa-source.c2
-rw-r--r--src/modules/alsa/alsa-util.c9
-rw-r--r--src/modules/alsa/alsa-util.h3
-rw-r--r--src/modules/bluetooth/bluetooth-util.c2
-rw-r--r--src/modules/bluetooth/module-bluetooth-device.c225
-rw-r--r--src/modules/module-null-sink.c7
-rw-r--r--src/modules/reserve-wrap.c11
-rw-r--r--src/modules/rtp/module-rtp-recv.c73
-rw-r--r--src/modules/rtp/module-rtp-send.c6
-rw-r--r--src/modules/rtp/rtp.c21
-rw-r--r--src/modules/rtp/rtp.h2
12 files changed, 283 insertions, 80 deletions
diff --git a/src/modules/alsa/alsa-sink.c b/src/modules/alsa/alsa-sink.c
index f9fb0335..c18c34ef 100644
--- a/src/modules/alsa/alsa-sink.c
+++ b/src/modules/alsa/alsa-sink.c
@@ -1728,7 +1728,7 @@ pa_sink *pa_alsa_sink_new(pa_module *m, pa_modargs *ma, const char*driver, pa_ca
if (setup_mixer(u, ignore_dB) < 0)
goto fail;
- pa_alsa_dump(u->pcm_handle);
+ pa_alsa_dump(PA_LOG_DEBUG, u->pcm_handle);
if (!(u->thread = pa_thread_new(thread_func, u))) {
pa_log("Failed to create thread.");
diff --git a/src/modules/alsa/alsa-source.c b/src/modules/alsa/alsa-source.c
index 9c36211b..9cbd79fa 100644
--- a/src/modules/alsa/alsa-source.c
+++ b/src/modules/alsa/alsa-source.c
@@ -1579,7 +1579,7 @@ pa_source *pa_alsa_source_new(pa_module *m, pa_modargs *ma, const char*driver, p
if (setup_mixer(u, ignore_dB) < 0)
goto fail;
- pa_alsa_dump(u->pcm_handle);
+ pa_alsa_dump(PA_LOG_DEBUG, u->pcm_handle);
if (!(u->thread = pa_thread_new(thread_func, u))) {
pa_log("Failed to create thread.");
diff --git a/src/modules/alsa/alsa-util.c b/src/modules/alsa/alsa-util.c
index 870cf0f1..fbf88b08 100644
--- a/src/modules/alsa/alsa-util.c
+++ b/src/modules/alsa/alsa-util.c
@@ -1297,7 +1297,7 @@ int pa_alsa_calc_mixer_map(snd_mixer_elem_t *elem, const pa_channel_map *channel
return 0;
}
-void pa_alsa_dump(snd_pcm_t *pcm) {
+void pa_alsa_dump(pa_log_level_t level, snd_pcm_t *pcm) {
int err;
snd_output_t *out;
@@ -1306,11 +1306,11 @@ void pa_alsa_dump(snd_pcm_t *pcm) {
pa_assert_se(snd_output_buffer_open(&out) == 0);
if ((err = snd_pcm_dump(pcm, out)) < 0)
- pa_log_debug("snd_pcm_dump(): %s", snd_strerror(err));
+ pa_logl(level, "snd_pcm_dump(): %s", snd_strerror(err));
else {
char *s = NULL;
snd_output_buffer_string(out, &s);
- pa_log_debug("snd_pcm_dump():\n%s", pa_strnull(s));
+ pa_logl(level, "snd_pcm_dump():\n%s", pa_strnull(s));
}
pa_assert_se(snd_output_close(out) == 0);
@@ -1612,6 +1612,7 @@ snd_pcm_sframes_t pa_alsa_safe_avail(snd_pcm_t *pcm, size_t hwbuf_size, const pa
(unsigned long) (pa_bytes_to_usec(k, ss) / PA_USEC_PER_MSEC),
pa_strnull(dn));
pa_xfree(dn);
+ pa_alsa_dump(PA_LOG_ERROR, pcm);
} PA_ONCE_END;
/* Mhmm, let's try not to fail completely */
@@ -1653,6 +1654,7 @@ int pa_alsa_safe_delay(snd_pcm_t *pcm, snd_pcm_sframes_t *delay, size_t hwbuf_si
(unsigned long) (pa_bytes_to_usec(abs_k, ss) / PA_USEC_PER_MSEC),
pa_strnull(dn));
pa_xfree(dn);
+ pa_alsa_dump(PA_LOG_ERROR, pcm);
} PA_ONCE_END;
/* Mhmm, let's try not to fail completely */
@@ -1698,6 +1700,7 @@ int pa_alsa_safe_mmap_begin(snd_pcm_t *pcm, const snd_pcm_channel_area_t **areas
(unsigned long) (pa_bytes_to_usec(k, ss) / PA_USEC_PER_MSEC),
pa_strnull(dn));
pa_xfree(dn);
+ pa_alsa_dump(PA_LOG_ERROR, pcm);
} PA_ONCE_END;
return r;
diff --git a/src/modules/alsa/alsa-util.h b/src/modules/alsa/alsa-util.h
index 94f27d14..9fce6daf 100644
--- a/src/modules/alsa/alsa-util.h
+++ b/src/modules/alsa/alsa-util.h
@@ -33,6 +33,7 @@
#include <pulsecore/rtpoll.h>
#include <pulsecore/core.h>
+#include <pulsecore/log.h>
typedef struct pa_alsa_fdlist pa_alsa_fdlist;
@@ -114,7 +115,7 @@ int pa_alsa_probe_profiles(
int pa_alsa_calc_mixer_map(snd_mixer_elem_t *elem, const pa_channel_map *channel_map, snd_mixer_selem_channel_id_t mixer_map[], pa_bool_t playback);
-void pa_alsa_dump(snd_pcm_t *pcm);
+void pa_alsa_dump(pa_log_level_t level, snd_pcm_t *pcm);
void pa_alsa_dump_status(snd_pcm_t *pcm);
void pa_alsa_redirect_errors_inc(void);
diff --git a/src/modules/bluetooth/bluetooth-util.c b/src/modules/bluetooth/bluetooth-util.c
index 6e4344f7..5c7681d4 100644
--- a/src/modules/bluetooth/bluetooth-util.c
+++ b/src/modules/bluetooth/bluetooth-util.c
@@ -773,6 +773,8 @@ void pa_bluetooth_discovery_unref(pa_bluetooth_discovery *y) {
if (y->core)
pa_shared_remove(y->core, "bluetooth-discovery");
+
+ pa_xfree(y);
}
void pa_bluetooth_discovery_sync(pa_bluetooth_discovery *y) {
diff --git a/src/modules/bluetooth/module-bluetooth-device.c b/src/modules/bluetooth/module-bluetooth-device.c
index 4613172e..90f64861 100644
--- a/src/modules/bluetooth/module-bluetooth-device.c
+++ b/src/modules/bluetooth/module-bluetooth-device.c
@@ -166,10 +166,14 @@ struct userdata {
pa_modargs *modargs;
- int stream_write_type, stream_read_type;
+ int stream_write_type;
int service_write_type, service_read_type;
};
+#define FIXED_LATENCY_PLAYBACK_A2DP (25*PA_USEC_PER_MSEC)
+#define FIXED_LATENCY_PLAYBACK_HSP (125*PA_USEC_PER_MSEC)
+#define FIXED_LATENCY_RECORD_HSP (25*PA_USEC_PER_MSEC)
+
#ifdef NOKIA
#define USE_SCO_OVER_PCM(u) (u->profile == PROFILE_HSP && (u->hsp.sco_sink && u->hsp.sco_source))
#endif
@@ -275,6 +279,7 @@ static ssize_t service_expect(struct userdata*u, bt_audio_msg_header_t *rsp, siz
return 0;
}
+/* Run from main thread */
static int parse_caps(struct userdata *u, uint8_t seid, const struct bt_get_capabilities_rsp *rsp) {
uint16_t bytes_left;
const codec_capabilities_t *codec;
@@ -335,6 +340,7 @@ static int parse_caps(struct userdata *u, uint8_t seid, const struct bt_get_capa
return 0;
}
+/* Run from main thread */
static int get_caps(struct userdata *u, uint8_t seid) {
union {
struct bt_get_capabilities_req getcaps_req;
@@ -374,6 +380,7 @@ static int get_caps(struct userdata *u, uint8_t seid) {
return get_caps(u, ret);
}
+/* Run from main thread */
static uint8_t a2dp_default_bitpool(uint8_t freq, uint8_t mode) {
switch (freq) {
@@ -419,6 +426,7 @@ static uint8_t a2dp_default_bitpool(uint8_t freq, uint8_t mode) {
}
}
+/* Run from main thread */
static int setup_a2dp(struct userdata *u) {
sbc_capabilities_t *cap;
int i;
@@ -526,6 +534,7 @@ static int setup_a2dp(struct userdata *u) {
return 0;
}
+/* Run from main thread */
static void setup_sbc(struct a2dp_info *a2dp) {
sbc_capabilities_t *active_capabilities;
@@ -617,6 +626,7 @@ static void setup_sbc(struct a2dp_info *a2dp) {
a2dp->frame_length = sbc_get_frame_length(&a2dp->sbc);
}
+/* Run from main thread */
static int set_conf(struct userdata *u) {
union {
struct bt_open_req open_req;
@@ -705,6 +715,7 @@ static int start_stream_fd(struct userdata *u) {
uint8_t buf[BT_SUGGESTED_BUFFER_SIZE];
} msg;
struct pollfd *pollfd;
+ int one;
pa_assert(u);
pa_assert(u->rtpoll);
@@ -733,13 +744,29 @@ static int start_stream_fd(struct userdata *u) {
pa_make_fd_nonblock(u->stream_fd);
pa_make_socket_low_delay(u->stream_fd);
+ one = 1;
+ if (setsockopt(u->stream_fd, SOL_SOCKET, SO_TIMESTAMP, &one, sizeof(one)) < 0)
+ pa_log_warn("Failed to enable SO_TIMESTAMP: %s", pa_cstrerror(errno));
+
+ pa_log_debug("Stream properly set up, we're ready to roll!");
+
u->rtpoll_item = pa_rtpoll_item_new(u->rtpoll, PA_RTPOLL_NEVER, 1);
pollfd = pa_rtpoll_item_get_pollfd(u->rtpoll_item, NULL);
pollfd->fd = u->stream_fd;
pollfd->events = pollfd->revents = 0;
- u->read_index = 0;
- u->write_index = 0;
+ u->read_index = u->write_index = 0;
+ u->started_at = 0;
+
+ if (u->source)
+ u->read_smoother = pa_smoother_new(
+ PA_USEC_PER_SEC,
+ PA_USEC_PER_SEC*2,
+ TRUE,
+ TRUE,
+ 10,
+ pa_rtclock_usec(),
+ TRUE);
return 0;
}
@@ -775,9 +802,15 @@ static int stop_stream_fd(struct userdata *u) {
pa_close(u->stream_fd);
u->stream_fd = -1;
+ if (u->read_smoother) {
+ pa_smoother_free(u->read_smoother);
+ u->read_smoother = NULL;
+ }
+
return r;
}
+/* Run from IO thread */
static int sink_process_msg(pa_msgobject *o, int code, void *data, int64_t offset, pa_memchunk *chunk) {
struct userdata *u = PA_SINK(o)->userdata;
pa_bool_t failed = FALSE;
@@ -785,7 +818,6 @@ static int sink_process_msg(pa_msgobject *o, int code, void *data, int64_t offse
pa_assert(u->sink == PA_SINK(o));
- pa_log_debug("got message: %d", code);
switch (code) {
case PA_SINK_MESSAGE_SET_STATE:
@@ -813,8 +845,6 @@ static int sink_process_msg(pa_msgobject *o, int code, void *data, int64_t offse
if (!u->source || u->source->state == PA_SOURCE_SUSPENDED)
if (start_stream_fd(u) < 0)
failed = TRUE;
-
- u->started_at = pa_rtclock_usec();
break;
case PA_SINK_UNLINKED:
@@ -825,7 +855,24 @@ static int sink_process_msg(pa_msgobject *o, int code, void *data, int64_t offse
break;
case PA_SINK_MESSAGE_GET_LATENCY: {
- *((pa_usec_t*) data) = 0;
+
+ if (u->read_smoother) {
+ pa_usec_t wi, ri;
+
+ ri = pa_smoother_get(u->read_smoother, pa_rtclock_usec());
+ wi = pa_bytes_to_usec(u->write_index + u->block_size, &u->sample_spec);
+
+ *((pa_usec_t*) data) = wi > ri ? wi - ri : 0;
+ } else {
+ pa_usec_t ri, wi;
+
+ ri = pa_rtclock_usec() - u->started_at;
+ wi = pa_bytes_to_usec(u->write_index, &u->sample_spec);
+
+ *((pa_usec_t*) data) = wi > ri ? wi - ri : 0;
+ }
+
+ *((pa_usec_t*) data) += u->sink->fixed_latency;
return 0;
}
}
@@ -835,6 +882,7 @@ static int sink_process_msg(pa_msgobject *o, int code, void *data, int64_t offse
return (r < 0 || !failed) ? r : -1;
}
+/* Run from IO thread */
static int source_process_msg(pa_msgobject *o, int code, void *data, int64_t offset, pa_memchunk *chunk) {
struct userdata *u = PA_SOURCE(o)->userdata;
pa_bool_t failed = FALSE;
@@ -842,7 +890,6 @@ static int source_process_msg(pa_msgobject *o, int code, void *data, int64_t off
pa_assert(u->source == PA_SOURCE(o));
- pa_log_debug("got message: %d", code);
switch (code) {
case PA_SOURCE_MESSAGE_SET_STATE:
@@ -856,7 +903,8 @@ static int source_process_msg(pa_msgobject *o, int code, void *data, int64_t off
if (!u->sink || u->sink->state == PA_SINK_SUSPENDED)
stop_stream_fd(u);
- pa_smoother_pause(u->read_smoother, pa_rtclock_usec());
+ if (u->read_smoother)
+ pa_smoother_pause(u->read_smoother, pa_rtclock_usec());
break;
case PA_SOURCE_IDLE:
@@ -869,7 +917,8 @@ static int source_process_msg(pa_msgobject *o, int code, void *data, int64_t off
if (start_stream_fd(u) < 0)
failed = TRUE;
- pa_smoother_resume(u->read_smoother, pa_rtclock_usec(), TRUE);
+ /* We don't resume the smoother here. Instead we
+ * wait until the first packet arrives */
break;
case PA_SOURCE_UNLINKED:
@@ -880,7 +929,12 @@ static int source_process_msg(pa_msgobject *o, int code, void *data, int64_t off
break;
case PA_SOURCE_MESSAGE_GET_LATENCY: {
- *((pa_usec_t*) data) = 0;
+ pa_usec_t wi, ri;
+
+ wi = pa_smoother_get(u->read_smoother, pa_rtclock_usec());
+ ri = pa_bytes_to_usec(u->read_index, &u->sample_spec);
+
+ *((pa_usec_t*) data) = (wi > ri ? wi - ri : 0) + u->source->fixed_latency;
return 0;
}
@@ -891,6 +945,7 @@ static int source_process_msg(pa_msgobject *o, int code, void *data, int64_t off
return (r < 0 || !failed) ? r : -1;
}
+/* Run from IO thread */
static int hsp_process_render(struct userdata *u) {
int ret = 0;
@@ -947,12 +1002,14 @@ static int hsp_process_render(struct userdata *u) {
pa_memblock_unref(u->write_memchunk.memblock);
pa_memchunk_reset(&u->write_memchunk);
+ ret = 1;
break;
}
return ret;
}
+/* Run from IO thread */
static int hsp_process_push(struct userdata *u) {
int ret = 0;
pa_memchunk memchunk;
@@ -960,6 +1017,7 @@ static int hsp_process_push(struct userdata *u) {
pa_assert(u);
pa_assert(u->profile == PROFILE_HSP);
pa_assert(u->source);
+ pa_assert(u->read_smoother);
memchunk.memblock = pa_memblock_new(u->core->mempool, u->block_size);
memchunk.index = memchunk.length = 0;
@@ -967,9 +1025,26 @@ static int hsp_process_push(struct userdata *u) {
for (;;) {
ssize_t l;
void *p;
+ struct msghdr m;
+ struct cmsghdr *cm;
+ uint8_t aux[1024];
+ struct iovec iov;
+ pa_bool_t found_tstamp = FALSE;
+ pa_usec_t tstamp;
+
+ memset(&m, 0, sizeof(m));
+ memset(&aux, 0, sizeof(aux));
+ memset(&iov, 0, sizeof(iov));
+
+ m.msg_iov = &iov;
+ m.msg_iovlen = 1;
+ m.msg_control = aux;
+ m.msg_controllen = sizeof(aux);
p = pa_memblock_acquire(memchunk.memblock);
- l = pa_read(u->stream_fd, p, pa_memblock_get_length(memchunk.memblock), &u->stream_read_type);
+ iov.iov_base = p;
+ iov.iov_len = pa_memblock_get_length(memchunk.memblock);
+ l = recvmsg(u->stream_fd, &m, 0);
pa_memblock_release(memchunk.memblock);
if (l <= 0) {
@@ -992,7 +1067,26 @@ static int hsp_process_push(struct userdata *u) {
memchunk.length = (size_t) l;
u->read_index += (uint64_t) l;
+ for (cm = CMSG_FIRSTHDR(&m); cm; cm = CMSG_NXTHDR(&m, cm))
+ if (cm->cmsg_level == SOL_SOCKET && cm->cmsg_type == SO_TIMESTAMP) {
+ struct timeval *tv = (struct timeval*) CMSG_DATA(cm);
+ pa_rtclock_from_wallclock(tv);
+ tstamp = pa_timeval_load(tv);
+ found_tstamp = TRUE;
+ break;
+ }
+
+ if (!found_tstamp) {
+ pa_log_warn("Couldn't find SO_TIMESTAMP data in auxiliary recvmsg() data!");
+ tstamp = pa_rtclock_usec();
+ }
+
+ pa_smoother_put(u->read_smoother, tstamp, pa_bytes_to_usec(u->read_index, &u->sample_spec));
+ pa_smoother_resume(u->read_smoother, tstamp, TRUE);
+
pa_source_post(u->source, &memchunk);
+
+ ret = 1;
break;
}
@@ -1001,6 +1095,7 @@ static int hsp_process_push(struct userdata *u) {
return ret;
}
+/* Run from IO thread */
static void a2dp_prepare_buffer(struct userdata *u) {
pa_assert(u);
@@ -1012,6 +1107,7 @@ static void a2dp_prepare_buffer(struct userdata *u) {
u->a2dp.buffer = pa_xmalloc(u->a2dp.buffer_size);
}
+/* Run from IO thread */
static int a2dp_process_render(struct userdata *u) {
struct a2dp_info *a2dp;
struct rtp_header *header;
@@ -1095,7 +1191,7 @@ static int a2dp_process_render(struct userdata *u) {
header->v = 2;
header->pt = 1;
header->sequence_number = htons(a2dp->seq_num++);
- header->timestamp = htonl(u->write_index / pa_frame_size(&u->sink->sample_spec));
+ header->timestamp = htonl(u->write_index / pa_frame_size(&u->sample_spec));
header->ssrc = htonl(1);
payload->frame_count = frame_count;
@@ -1137,6 +1233,8 @@ static int a2dp_process_render(struct userdata *u) {
pa_memblock_unref(u->write_memchunk.memblock);
pa_memchunk_reset(&u->write_memchunk);
+ ret = 1;
+
break;
}
@@ -1145,7 +1243,8 @@ static int a2dp_process_render(struct userdata *u) {
static void thread_func(void *userdata) {
struct userdata *u = userdata;
- pa_bool_t do_write = FALSE, writable = FALSE;
+ unsigned do_write = 0;
+ pa_bool_t writable = FALSE;
pa_assert(u);
@@ -1160,8 +1259,6 @@ static void thread_func(void *userdata) {
pa_thread_mq_install(&u->thread_mq);
pa_rtpoll_install(u->rtpoll);
- pa_smoother_set_time_offset(u->read_smoother, pa_rtclock_usec());
-
for (;;) {
struct pollfd *pollfd;
int ret;
@@ -1171,13 +1268,20 @@ static void thread_func(void *userdata) {
if (u->source && PA_SOURCE_IS_LINKED(u->source->thread_info.state)) {
+ /* We should send two blocks to the device before we expect
+ * a response. */
+
+ if (u->write_index == 0 && u->read_index <= 0)
+ do_write = 2;
+
if (pollfd && (pollfd->revents & POLLIN)) {
+ int n_read;
- if (hsp_process_push(u) < 0)
+ if ((n_read = hsp_process_push(u)) < 0)
goto fail;
/* We just read something, so we are supposed to write something, too */
- do_write = TRUE;
+ do_write += n_read;
}
}
@@ -1190,7 +1294,7 @@ static void thread_func(void *userdata) {
if (pollfd->revents & POLLOUT)
writable = TRUE;
- if ((!u->source || !PA_SOURCE_IS_LINKED(u->source->thread_info.state)) && !do_write && writable) {
+ if ((!u->source || !PA_SOURCE_IS_LINKED(u->source->thread_info.state)) && do_write <= 0 && writable) {
pa_usec_t time_passed;
uint64_t should_have_written;
@@ -1198,36 +1302,37 @@ static void thread_func(void *userdata) {
* to. So let's do things by time */
time_passed = pa_rtclock_usec() - u->started_at;
- should_have_written = pa_usec_to_bytes(time_passed, &u->sink->sample_spec);
+ should_have_written = pa_usec_to_bytes(time_passed, &u->sample_spec);
- do_write = u->write_index <= should_have_written ;
-/* pa_log_debug("Time has come: %s", pa_yes_no(do_write)); */
+ do_write = u->write_index <= should_have_written;
}
- if (writable && do_write) {
- if (u->write_index == 0)
+ if (writable && do_write > 0) {
+ int n_written;
+
+ if (u->write_index <= 0)
u->started_at = pa_rtclock_usec();
if (u->profile == PROFILE_A2DP) {
- if (a2dp_process_render(u) < 0)
+ if ((n_written = a2dp_process_render(u)) < 0)
goto fail;
} else {
- if (hsp_process_render(u) < 0)
+ if ((n_written = hsp_process_render(u)) < 0)
goto fail;
}
- do_write = FALSE;
+ do_write -= n_written;
writable = FALSE;
}
- if ((!u->source || !PA_SOURCE_IS_LINKED(u->source->thread_info.state)) && !do_write) {
+ if ((!u->source || !PA_SOURCE_IS_LINKED(u->source->thread_info.state)) && do_write <= 0) {
pa_usec_t time_passed, next_write_at, sleep_for;
/* Hmm, there is no input stream we could synchronize
* to. So let's estimate when we need to wake up the latest */
time_passed = pa_rtclock_usec() - u->started_at;
- next_write_at = pa_bytes_to_usec(u->write_index, &u->sink->sample_spec);
+ next_write_at = pa_bytes_to_usec(u->write_index, &u->sample_spec);
sleep_for = time_passed < next_write_at ? next_write_at - time_passed : 0;
/* pa_log("Sleeping for %lu; time passed %lu, next write at %lu", (unsigned long) sleep_for, (unsigned long) time_passed, (unsigned long)next_write_at); */
@@ -1255,7 +1360,11 @@ static void thread_func(void *userdata) {
pollfd = u->rtpoll_item ? pa_rtpoll_item_get_pollfd(u->rtpoll_item, NULL) : NULL;
if (pollfd && (pollfd->revents & ~(POLLOUT|POLLIN))) {
- pa_log_error("FD error.");
+ pa_log_info("FD error: %s%s%s%s",
+ pollfd->revents & POLLERR ? "POLLERR " :"",
+ pollfd->revents & POLLHUP ? "POLLHUP " :"",
+ pollfd->revents & POLLPRI ? "POLLPRI " :"",
+ pollfd->revents & POLLNVAL ? "POLLNVAL " :"");
goto fail;
}
}
@@ -1270,6 +1379,7 @@ finish:
pa_log_debug("IO thread shutting down");
}
+/* Run from main thread */
static DBusHandlerResult filter_cb(DBusConnection *bus, DBusMessage *m, void *userdata) {
DBusError err;
struct userdata *u;
@@ -1302,12 +1412,12 @@ static DBusHandlerResult filter_cb(DBusConnection *bus, DBusMessage *m, void *us
if (u->profile == PROFILE_HSP) {
if (u->sink && dbus_message_is_signal(m, "org.bluez.Headset", "SpeakerGainChanged")) {
- pa_cvolume_set(&v, u->sink->sample_spec.channels, (pa_volume_t) (gain * PA_VOLUME_NORM / 15));
+ pa_cvolume_set(&v, u->sample_spec.channels, (pa_volume_t) (gain * PA_VOLUME_NORM / 15));
pa_sink_volume_changed(u->sink, &v);
} else if (u->source && dbus_message_is_signal(m, "org.bluez.Headset", "MicrophoneGainChanged")) {
- pa_cvolume_set(&v, u->sink->sample_spec.channels, (pa_volume_t) (gain * PA_VOLUME_NORM / 15));
+ pa_cvolume_set(&v, u->sample_spec.channels, (pa_volume_t) (gain * PA_VOLUME_NORM / 15));
pa_source_volume_changed(u->source, &v);
}
}
@@ -1319,6 +1429,7 @@ fail:
return DBUS_HANDLER_RESULT_NOT_YET_HANDLED;
}
+/* Run from main thread */
static void sink_set_volume_cb(pa_sink *s) {
struct userdata *u = s->userdata;
DBusMessage *m;
@@ -1334,7 +1445,7 @@ static void sink_set_volume_cb(pa_sink *s) {
if (gain > 15)
gain = 15;
- pa_cvolume_set(&s->virtual_volume, u->sink->sample_spec.channels, (pa_volume_t) (gain * PA_VOLUME_NORM / 15));
+ pa_cvolume_set(&s->virtual_volume, u->sample_spec.channels, (pa_volume_t) (gain * PA_VOLUME_NORM / 15));
pa_assert_se(m = dbus_message_new_method_call("org.bluez", u->path, "org.bluez.Headset", "SetSpeakerGain"));
pa_assert_se(dbus_message_append_args(m, DBUS_TYPE_UINT16, &gain, DBUS_TYPE_INVALID));
@@ -1342,6 +1453,7 @@ static void sink_set_volume_cb(pa_sink *s) {
dbus_message_unref(m);
}
+/* Run from main thread */
static void source_set_volume_cb(pa_source *s) {
struct userdata *u = s->userdata;
DBusMessage *m;
@@ -1357,7 +1469,7 @@ static void source_set_volume_cb(pa_source *s) {
if (gain > 15)
gain = 15;
- pa_cvolume_set(&s->virtual_volume, u->source->sample_spec.channels, (pa_volume_t) (gain * PA_VOLUME_NORM / 15));
+ pa_cvolume_set(&s->virtual_volume, u->sample_spec.channels, (pa_volume_t) (gain * PA_VOLUME_NORM / 15));
pa_assert_se(m = dbus_message_new_method_call("org.bluez", u->path, "org.bluez.Headset", "SetMicrophoneGain"));
pa_assert_se(dbus_message_append_args(m, DBUS_TYPE_UINT16, &gain, DBUS_TYPE_INVALID));
@@ -1365,6 +1477,7 @@ static void source_set_volume_cb(pa_source *s) {
dbus_message_unref(m);
}
+/* Run from main thread */
static char *get_name(const char *type, pa_modargs *ma, const char *device_id, pa_bool_t *namereg_fail) {
char *t;
const char *n;
@@ -1451,6 +1564,7 @@ static pa_hook_result_t source_state_changed_cb(pa_core *c, pa_source *s, struct
#endif
+/* Run from main thread */
static int add_sink(struct userdata *u) {
#ifdef NOKIA
@@ -1492,6 +1606,11 @@ static int add_sink(struct userdata *u) {
u->sink->userdata = u;
u->sink->parent.process_msg = sink_process_msg;
+
+ pa_sink_set_max_request(u->sink, u->block_size);
+ u->sink->fixed_latency =
+ (u->profile == PROFILE_A2DP ? FIXED_LATENCY_PLAYBACK_A2DP : FIXED_LATENCY_PLAYBACK_HSP) +
+ pa_bytes_to_usec(u->block_size, &u->sample_spec);
}
if (u->profile == PROFILE_HSP) {
@@ -1502,12 +1621,13 @@ static int add_sink(struct userdata *u) {
return 0;
}
+/* Run from main thread */
static int add_source(struct userdata *u) {
#ifdef NOKIA
if (USE_SCO_OVER_PCM(u)) {
u->source = u->hsp.sco_source;
- pa_proplist_sets(u->source->proplist, "bluetooth.protocol", "sco");
+ pa_proplist_sets(u->source->proplist, "bluetooth.protocol", "hsp");
if (!u->hsp.source_state_changed_slot)
u->hsp.source_state_changed_slot = pa_hook_connect(&u->core->hooks[PA_CORE_HOOK_SOURCE_STATE_CHANGED], PA_HOOK_NORMAL, (pa_hook_cb_t) source_state_changed_cb, u);
@@ -1523,7 +1643,7 @@ static int add_source(struct userdata *u) {
data.driver = __FILE__;
data.module = u->module;
pa_source_new_data_set_sample_spec(&data, &u->sample_spec);
- pa_proplist_sets(data.proplist, "bluetooth.protocol", u->profile == PROFILE_A2DP ? "a2dp" : "sco");
+ pa_proplist_sets(data.proplist, "bluetooth.protocol", u->profile == PROFILE_A2DP ? "a2dp" : "hsp");
data.card = u->card;
data.name = get_name("source", u->modargs, u->address, &b);
data.namereg_fail = b;
@@ -1538,6 +1658,10 @@ static int add_source(struct userdata *u) {
u->source->userdata = u;
u->source->parent.process_msg = source_process_msg;
+
+ u->source->fixed_latency =
+ (/* u->profile == PROFILE_A2DP ? FIXED_LATENCY_RECORD_A2DP : */ FIXED_LATENCY_RECORD_HSP) +
+ pa_bytes_to_usec(u->block_size, &u->sample_spec);
}
if (u->profile == PROFILE_HSP) {
@@ -1549,6 +1673,7 @@ static int add_source(struct userdata *u) {
return 0;
}
+/* Run from main thread */
static void shutdown_bt(struct userdata *u) {
pa_assert(u);
@@ -1557,12 +1682,12 @@ static void shutdown_bt(struct userdata *u) {
u->stream_fd = -1;
u->stream_write_type = 0;
- u->stream_read_type = 0;
}
if (u->service_fd >= 0) {
pa_close(u->service_fd);
u->service_fd = -1;
+ u->service_write_type = u->service_write_type = 0;
}
if (u->write_memchunk.memblock) {
@@ -1571,12 +1696,13 @@ static void shutdown_bt(struct userdata *u) {
}
}
+/* Run from main thread */
static int init_bt(struct userdata *u) {
pa_assert(u);
shutdown_bt(u);
- u->stream_write_type = u->stream_read_type = 0;
+ u->stream_write_type = 0;
u->service_write_type = u->service_write_type = 0;
if ((u->service_fd = bt_audio_service_open()) < 0) {
@@ -1589,6 +1715,7 @@ static int init_bt(struct userdata *u) {
return 0;
}
+/* Run from main thread */
static int setup_bt(struct userdata *u) {
pa_assert(u);
@@ -1614,6 +1741,7 @@ static int setup_bt(struct userdata *u) {
return 0;
}
+/* Run from main thread */
static int init_profile(struct userdata *u) {
int r = 0;
pa_assert(u);
@@ -1634,6 +1762,7 @@ static int init_profile(struct userdata *u) {
return r;
}
+/* Run from main thread */
static void stop_thread(struct userdata *u) {
pa_assert(u);
@@ -1674,8 +1803,14 @@ static void stop_thread(struct userdata *u) {
pa_rtpoll_free(u->rtpoll);
u->rtpoll = NULL;
}
+
+ if (u->read_smoother) {
+ pa_smoother_free(u->read_smoother);
+ u->read_smoother = NULL;
+ }
}
+/* Run from main thread */
static int start_thread(struct userdata *u) {
pa_assert(u);
pa_assert(!u->thread);
@@ -1724,6 +1859,7 @@ static int start_thread(struct userdata *u) {
return 0;
}
+/* Run from main thread */
static int card_set_profile(pa_card *c, pa_card_profile *new_profile) {
struct userdata *u;
enum profile *d;
@@ -1797,6 +1933,7 @@ static int card_set_profile(pa_card *c, pa_card_profile *new_profile) {
return 0;
}
+/* Run from main thread */
static int add_card(struct userdata *u, const char *default_profile, const pa_bluetooth_device *device) {
pa_card_new_data data;
pa_bool_t b;
@@ -1845,7 +1982,7 @@ static int add_card(struct userdata *u, const char *default_profile, const pa_bl
}
if (pa_bluetooth_uuid_has(device->uuids, HSP_HS_UUID) ||
- pa_bluetooth_uuid_has(device->uuids, HFP_HS_UUID)) {
+ pa_bluetooth_uuid_has(device->uuids, HFP_HS_UUID)) {
p = pa_card_profile_new("hsp", _("Telephony Duplex (HSP/HFP)"), sizeof(enum profile));
p->priority = 20;
p->n_sinks = 1;
@@ -1890,6 +2027,7 @@ static int add_card(struct userdata *u, const char *default_profile, const pa_bl
return 0;
}
+/* Run from main thread */
static const pa_bluetooth_device* find_device(struct userdata *u, const char *address, const char *path) {
const pa_bluetooth_device *d = NULL;
@@ -1926,6 +2064,7 @@ static const pa_bluetooth_device* find_device(struct userdata *u, const char *ad
return d;
}
+/* Run from main thread */
static int setup_dbus(struct userdata *u) {
DBusError err;
@@ -1965,14 +2104,6 @@ int pa__init(pa_module* m) {
u->core = m->core;
u->service_fd = -1;
u->stream_fd = -1;
- u->read_smoother = pa_smoother_new(
- PA_USEC_PER_SEC,
- PA_USEC_PER_SEC*2,
- TRUE,
- TRUE,
- 10,
- 0,
- FALSE);
u->sample_spec = m->core->default_sample_spec;
u->modargs = ma;
diff --git a/src/modules/module-null-sink.c b/src/modules/module-null-sink.c
index 129bc1c3..d9bab6bd 100644
--- a/src/modules/module-null-sink.c
+++ b/src/modules/module-null-sink.c
@@ -119,6 +119,7 @@ static int sink_process_msg(
static void sink_update_requested_latency_cb(pa_sink *s) {
struct userdata *u;
+ size_t nbytes;
pa_sink_assert_ref(s);
pa_assert_se(u = s->userdata);
@@ -127,6 +128,10 @@ static void sink_update_requested_latency_cb(pa_sink *s) {
if (u->block_usec == (pa_usec_t) -1)
u->block_usec = s->thread_info.max_latency;
+
+ nbytes = pa_usec_to_bytes(u->block_usec, &s->sample_spec);
+ pa_sink_set_max_rewind_within_thread(s, nbytes);
+ pa_sink_set_max_request_within_thread(s, nbytes);
}
static void process_rewind(struct userdata *u, pa_usec_t now) {
@@ -284,7 +289,7 @@ int pa__init(pa_module*m) {
pa_proplist_sets(data.proplist, PA_PROP_DEVICE_DESCRIPTION, pa_modargs_get_value(ma, "description", "Null Output"));
pa_proplist_sets(data.proplist, PA_PROP_DEVICE_CLASS, "abstract");
- u->sink = pa_sink_new(m->core, &data, PA_SINK_LATENCY);
+ u->sink = pa_sink_new(m->core, &data, PA_SINK_LATENCY|PA_SINK_DYNAMIC_LATENCY);
pa_sink_new_data_done(&data);
if (!u->sink) {
diff --git a/src/modules/reserve-wrap.c b/src/modules/reserve-wrap.c
index 1927342b..02ff29be 100644
--- a/src/modules/reserve-wrap.c
+++ b/src/modules/reserve-wrap.c
@@ -23,6 +23,8 @@
#include <config.h>
#endif
+#include <errno.h>
+
#include <pulse/xmalloc.h>
#include <pulse/i18n.h>
@@ -127,8 +129,13 @@ pa_reserve_wrapper* pa_reserve_wrapper_get(pa_core *c, const char *device_name)
request_cb,
NULL)) < 0) {
- pa_log_error("Failed to acquire reservation lock on device '%s': %s", device_name, pa_cstrerror(-k));
- goto fail;
+ if (k == -EBUSY) {
+ pa_log_error("Device '%s' already locked.", device_name);
+ goto fail;
+ } else {
+ pa_log_warn("Failed to acquire reservation lock on device '%s': %s", device_name, pa_cstrerror(-k));
+ return r;
+ }
}
pa_log_debug("Successfully acquired reservation lock on device '%s'", device_name);
diff --git a/src/modules/rtp/module-rtp-recv.c b/src/modules/rtp/module-rtp-recv.c
index e7749cdd..c61d2d8b 100644
--- a/src/modules/rtp/module-rtp-recv.c
+++ b/src/modules/rtp/module-rtp-recv.c
@@ -52,6 +52,8 @@
#include <pulsecore/rtclock.h>
#include <pulsecore/atomic.h>
#include <pulsecore/time-smoother.h>
+#include <pulsecore/socket-util.h>
+#include <pulsecore/once.h>
#include "module-rtp-recv-symdef.h"
@@ -165,7 +167,7 @@ static void sink_input_process_rewind_cb(pa_sink_input *i, size_t nbytes) {
pa_memblockq_rewind(s->memblockq, nbytes);
}
-/* Called from thread context */
+/* Called from I/O thread context */
static void sink_input_update_max_rewind_cb(pa_sink_input *i, size_t nbytes) {
struct session *s;
@@ -184,11 +186,24 @@ static void sink_input_kill(pa_sink_input* i) {
session_free(s);
}
+/* Called from IO context */
+static void sink_input_suspend_within_thread(pa_sink_input* i, pa_bool_t b) {
+ struct session *s;
+ pa_sink_input_assert_ref(i);
+ pa_assert_se(s = i->userdata);
+
+ if (b) {
+ pa_smoother_pause(s->smoother, pa_rtclock_usec());
+ pa_memblockq_flush_read(s->memblockq);
+ } else
+ s->first_packet = FALSE;
+}
+
/* Called from I/O thread context */
static int rtpoll_work_cb(pa_rtpoll_item *i) {
pa_memchunk chunk;
int64_t k, j, delta;
- struct timeval now;
+ struct timeval now = { 0, 0 };
struct session *s;
struct pollfd *p;
@@ -206,10 +221,11 @@ static int rtpoll_work_cb(pa_rtpoll_item *i) {
p->revents = 0;
- if (pa_rtp_recv(&s->rtp_context, &chunk, s->userdata->module->core->mempool) < 0)
+ if (pa_rtp_recv(&s->rtp_context, &chunk, s->userdata->module->core->mempool, &now) < 0)
return 0;
- if (s->sdp_info.payload != s->rtp_context.payload) {
+ if (s->sdp_info.payload != s->rtp_context.payload ||
+ !PA_SINK_IS_OPENED(s->sink_input->sink->thread_info.state)) {
pa_memblock_unref(chunk.memblock);
return 0;
}
@@ -240,10 +256,19 @@ static int rtpoll_work_cb(pa_rtpoll_item *i) {
pa_memblockq_seek(s->memblockq, delta * (int64_t) s->rtp_context.frame_size, PA_SEEK_RELATIVE, TRUE);
- pa_rtclock_get(&now);
+ if (now.tv_sec == 0) {
+ PA_ONCE_BEGIN {
+ pa_log_warn("Using artificial time instead of timestamp");
+ } PA_ONCE_END;
+ pa_rtclock_get(&now);
+ } else
+ pa_rtclock_from_wallclock(&now);
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));
+ /* Tell the smoother that we are rolling now, in case it is still paused */
+ pa_smoother_resume(s->smoother, pa_timeval_load(&now), TRUE);
+
if (pa_memblockq_push(s->memblockq, &chunk) < 0) {
pa_log_warn("Queue overrun");
pa_memblockq_seek(s->memblockq, (int64_t) chunk.length, PA_SEEK_RELATIVE, TRUE);
@@ -262,14 +287,14 @@ static int rtpoll_work_cb(pa_rtpoll_item *i) {
pa_usec_t wi, ri, render_delay, sink_delay = 0, latency, fix;
unsigned fix_samples;
- pa_log("Updating sample rate");
+ pa_log_debug("Updating sample rate");
wi = pa_smoother_get(s->smoother, pa_timeval_load(&now));
ri = pa_bytes_to_usec((uint64_t) pa_memblockq_get_read_index(s->memblockq), &s->sink_input->sample_spec);
- if (PA_MSGOBJECT(s->sink_input->sink)->process_msg(PA_MSGOBJECT(s->sink_input->sink), PA_SINK_MESSAGE_GET_LATENCY, &sink_delay, 0, NULL) < 0)
- sink_delay = 0;
+ pa_log_debug("wi=%lu ri=%lu", (unsigned long) wi, (unsigned long) ri);
+ sink_delay = pa_sink_get_latency_within_thread(s->sink_input->sink);
render_delay = pa_bytes_to_usec(pa_memblockq_get_length(s->sink_input->thread_info.render_memblockq), &s->sink_input->sink->sample_spec);
if (ri > render_delay+sink_delay)
@@ -294,14 +319,20 @@ static int rtpoll_work_cb(pa_rtpoll_item *i) {
fix_samples = (unsigned) (fix * (pa_usec_t) s->sink_input->thread_info.sample_spec.rate / (pa_usec_t) RATE_UPDATE_INTERVAL);
/* Check if deviation is in bounds */
- if (fix_samples > s->sink_input->sample_spec.rate*.20)
+ if (fix_samples > s->sink_input->sample_spec.rate*.50)
pa_log_debug("Hmmm, rate fix is too large (%lu Hz), not applying.", (unsigned long) fix_samples);
+ else {
+ /* Fix up rate */
+ if (latency < s->intended_latency)
+ s->sink_input->sample_spec.rate -= fix_samples;
+ else
+ s->sink_input->sample_spec.rate += fix_samples;
+
+ if (s->sink_input->sample_spec.rate > PA_RATE_MAX)
+ s->sink_input->sample_spec.rate = PA_RATE_MAX;
+ }
- /* Fix up rate */
- if (latency < s->intended_latency)
- s->sink_input->sample_spec.rate -= fix_samples;
- else
- s->sink_input->sample_spec.rate += fix_samples;
+ pa_assert(pa_sample_spec_valid(&s->sink_input->sample_spec));
pa_resampler_set_input_rate(s->sink_input->thread_info.resampler, s->sink_input->sample_spec.rate);
@@ -362,6 +393,14 @@ static int mcast_socket(const struct sockaddr* sa, socklen_t salen) {
goto fail;
}
+ pa_make_udp_socket_low_delay(fd);
+
+ one = 1;
+ if (setsockopt(fd, SOL_SOCKET, SO_TIMESTAMP, &one, sizeof(one)) < 0) {
+ pa_log("SO_TIMESTAMP failed: %s", pa_cstrerror(errno));
+ goto fail;
+ }
+
one = 1;
if (setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &one, sizeof(one)) < 0) {
pa_log("SO_REUSEADDR failed: %s", pa_cstrerror(errno));
@@ -437,7 +476,7 @@ static struct session *session_new(struct userdata *u, const pa_sdp_info *sdp_in
TRUE,
10,
pa_timeval_load(&now),
- FALSE);
+ TRUE);
s->last_rate_update = pa_timeval_load(&now);
pa_atomic_store(&s->timestamp, (int) now.tv_sec);
@@ -478,6 +517,7 @@ static struct session *session_new(struct userdata *u, const pa_sdp_info *sdp_in
s->sink_input->kill = sink_input_kill;
s->sink_input->attach = sink_input_attach;
s->sink_input->detach = sink_input_detach;
+ s->sink_input->suspend_within_thread = sink_input_suspend_within_thread;
pa_sink_input_get_silence(s->sink_input, &silence);
@@ -654,8 +694,7 @@ int pa__init(pa_module*m) {
if ((fd = mcast_socket(sa, salen)) < 0)
goto fail;
- u = pa_xnew(struct userdata, 1);
- m->userdata = u;
+ m->userdata = u = pa_xnew(struct userdata, 1);
u->module = m;
u->sink_name = pa_xstrdup(pa_modargs_get_value(ma, "sink", NULL));
diff --git a/src/modules/rtp/module-rtp-send.c b/src/modules/rtp/module-rtp-send.c
index 722d12bd..cdd2c57d 100644
--- a/src/modules/rtp/module-rtp-send.c
+++ b/src/modules/rtp/module-rtp-send.c
@@ -347,10 +347,10 @@ int pa__init(pa_module*m) {
o->push = source_output_push;
o->kill = source_output_kill;
- u = pa_xnew(struct userdata, 1);
- m->userdata = u;
- o->userdata = u;
+ pa_log_info("Configured source latency of %lu ms.",
+ pa_source_output_set_requested_latency(o, pa_bytes_to_usec(mtu, &o->sample_spec)) / PA_USEC_PER_MSEC);
+ m->userdata = o->userdata = u = pa_xnew(struct userdata, 1);
u->module = m;
u->source_output = o;
diff --git a/src/modules/rtp/rtp.c b/src/modules/rtp/rtp.c
index 7537c1f5..6706a10f 100644
--- a/src/modules/rtp/rtp.c
+++ b/src/modules/rtp/rtp.c
@@ -162,13 +162,16 @@ pa_rtp_context* pa_rtp_context_init_recv(pa_rtp_context *c, int fd, size_t frame
return c;
}
-int pa_rtp_recv(pa_rtp_context *c, pa_memchunk *chunk, pa_mempool *pool) {
+int pa_rtp_recv(pa_rtp_context *c, pa_memchunk *chunk, pa_mempool *pool, struct timeval *tstamp) {
int size;
struct msghdr m;
+ struct cmsghdr *cm;
struct iovec iov;
uint32_t header;
unsigned cc;
ssize_t r;
+ uint8_t aux[1024];
+ pa_bool_t found_tstamp = FALSE;
pa_assert(c);
pa_assert(chunk);
@@ -208,8 +211,8 @@ int pa_rtp_recv(pa_rtp_context *c, pa_memchunk *chunk, pa_mempool *pool) {
m.msg_namelen = 0;
m.msg_iov = &iov;
m.msg_iovlen = 1;
- m.msg_control = NULL;
- m.msg_controllen = 0;
+ m.msg_control = aux;
+ m.msg_controllen = sizeof(aux);
m.msg_flags = 0;
r = recvmsg(c->fd, &m, 0);
@@ -275,6 +278,18 @@ int pa_rtp_recv(pa_rtp_context *c, pa_memchunk *chunk, pa_mempool *pool) {
pa_memchunk_reset(&c->memchunk);
}
+ for (cm = CMSG_FIRSTHDR(&m); cm; cm = CMSG_NXTHDR(&m, cm)) {
+ if (cm->cmsg_level == SOL_SOCKET && cm->cmsg_type == SO_TIMESTAMP)
+ memcpy(tstamp, CMSG_DATA(cm), sizeof(struct timeval));
+ found_tstamp = TRUE;
+ break;
+ }
+
+ if (!found_tstamp) {
+ pa_log_warn("Couldn't find SO_TIMESTAMP data in auxiliary recvmsg() data!");
+ memset(tstamp, 0, sizeof(tstamp));
+ }
+
return 0;
fail:
diff --git a/src/modules/rtp/rtp.h b/src/modules/rtp/rtp.h
index eff5e75b..b197e82f 100644
--- a/src/modules/rtp/rtp.h
+++ b/src/modules/rtp/rtp.h
@@ -43,7 +43,7 @@ pa_rtp_context* pa_rtp_context_init_send(pa_rtp_context *c, int fd, uint32_t ssr
int pa_rtp_send(pa_rtp_context *c, size_t size, pa_memblockq *q);
pa_rtp_context* pa_rtp_context_init_recv(pa_rtp_context *c, int fd, size_t frame_size);
-int pa_rtp_recv(pa_rtp_context *c, pa_memchunk *chunk, pa_mempool *pool);
+int pa_rtp_recv(pa_rtp_context *c, pa_memchunk *chunk, pa_mempool *pool, struct timeval *tstamp);
void pa_rtp_context_destroy(pa_rtp_context *c);
generated by cgit (git 2.34.1) at 2024-05-18 12:36:59 +0000